OXYGENHUB

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

811 of 973 lines run, 143 of 219 functions · run #316 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
}
60
61
/// A tag as a remote reports it (spec §10.1).
62
#[derive(Clone, Debug, PartialEq, Eq)]
63
pub struct TagRef {
64
    pub name: String,
65
    pub state: ObjectRef,
66
    pub heads: Vec<ChangeHash>,
67
    pub attestation: Option<SignedAttest>,
68
}
69
70
/// One reference as a remote reports it.
71
#[derive(Clone, Debug)]
72
pub struct RemoteRef {
73
    pub name: String,
74
    pub state: ObjectRef,
75
    pub heads: Vec<ChangeHash>,
76
    /// The latest attestation, only when it describes `state`. An unattested
77
    /// push (accepted while no trust policy exists) leaves it `None`.
78
    pub attestation: Option<SignedAttest>,
79
    /// Serial of the latest attestation ever recorded, stale or not, so the
80
    /// next attestation continues the sequence.
81
    pub last_serial: Option<u64>,
82
    /// The host's acceptance of `attestation` when a short-lived key signed
83
    /// it (spec §14.7): the time its certificate is checked at.
84
    pub accepted: Option<SignedAttest>,
85
}
86
87
/// Everything needed to add a set of changes to a repository.
88
#[derive(Clone, Debug, Default)]
89
pub struct Bundle {
90
    /// In dependency order, each with its individual signatures.
91
    pub changes: Vec<(Change, Vec<Signature>)>,
92
    /// Seals covering the changes.
93
    pub attests: Vec<SignedAttest>,
94
    /// Chunks, blobs, identities, and state objects referenced by the changes.
95
    pub objects: Vec<(ObjectType, Vec<u8>)>,
96
    /// Policies between the receiver's and the one sent with the push,
97
    /// ascending by serial (spec §14.5).
98
    pub policies: Vec<SignedPolicy>,
99
    /// Each change's hash, in the order of `changes`, when whoever made
100
    /// the bundle knows it: the store it came from, or the bytes it was
101
    /// decoded from. Never read from the wire. Empty when unknown.
102
    pub hashes: Vec<ChangeHash>,
103
    /// Changes as stored, with their signatures, in dependency order, for a
104
    /// fetch between repositories on one machine: their text blobs and
105
    /// chunks are among `objects`. Never on the wire.
106
    pub raw_changes: Vec<(Vec<u8>, Vec<Signature>)>,
107
    /// Certificates of short-lived keys that signed what the bundle carries
108
    /// (spec §14.7), their issuers' identities among `objects`.
109
    pub certificates: Vec<crate::cert::SignedCertificate>,
110
}
111
112
impl Bundle {
11327×
    pub fn is_empty(&self) -> bool {
11427×
        self.changes.is_empty() && self.attests.is_empty() && self.objects.is_empty()
11527×
    }
116
}
117
118
/// A requested bookmark move (spec §10.2, §16.3).
119
#[derive(Clone, Debug)]
120
pub struct RefUpdate {
121
    pub name: String,
122
    pub old_state: ObjectRef,
123
    pub new_state: ObjectRef,
124
    pub force: bool,
125
    pub attestation: Option<SignedAttest>,
126
}
127
128
#[derive(Clone, Debug, PartialEq, Eq)]
129
pub struct RefResult {
130
    pub name: String,
131
    pub ok: bool,
132
    pub reason: String,
133
}
134
135
pub trait Remote {
136
    fn repo_id(&self) -> Result<[u8; 16]>;
137
    fn refs(&self) -> Result<RefsResponse>;
138
    /// Changes in the closure of `want` but not of `have`, in dependency order.
139
    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>>;
140
    /// The changes with their signatures, seals, identities, and blobs; blobs
141
    /// larger than `max_blob` bytes are left out for lazy fetching (spec §13.5).
142
    fn fetch(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle>;
143
    /// Individual objects by reference (spec §16.1 `objects`); a blob comes
144
    /// with its chunks.
145
    fn objects(&self, refs: &[ObjectRef]) -> Result<Vec<(ObjectType, Vec<u8>)>>;
146
    /// A checkpoint of exactly `heads` (spec §16.1 `checkpoint`): the
147
    /// checkpoint object, its payload blob, and the blob's chunks.
148
    fn checkpoint(&self, heads: &[ChangeHash]) -> Result<Bundle>;
149
    /// The changes within `depth` dependency steps of `heads` (spec §17).
150
    fn shallow(&self, heads: &[ChangeHash], depth: u32) -> Result<Vec<ChangeHash>>;
151
    /// Push changes, bookmark updates, new tags, and optionally a newer
152
    /// policy. The policy is applied first; a result named `policy` reports
153
    /// it, and each tag reports as `tags/<name>`.
154
    fn push(
155
        &self,
156
        bundle: Bundle,
157
        updates: Vec<RefUpdate>,
158
        policy: Option<SignedPolicy>,
159
        tags: Vec<TagRef>,
160
    ) -> Result<Vec<RefResult>>;
161
}
162
163
// ---------------------------------------------------------------- local
164
165
/// Another repository on this machine, served in-process.
166
pub struct LocalRemote {
167
    pub repo: Repo,
168
    /// The pushing repository's id, which binds an unbound target on the
169
    /// first push, as `Oxy-Repo-Id` does over HTTP.
170
    pub claimed_id: Option<[u8; 16]>,
171
}
172
173
impl LocalRemote {
1741063×
    pub fn open(path: &Path) -> Result<LocalRemote> {
175
        Ok(LocalRemote {
1761063×
            repo: Repo::open(path)?,
1771063×
            claimed_id: None,
178
        })
1791063×
    }
180
}
181
182
impl Remote for LocalRemote {
183836×
    fn repo_id(&self) -> Result<[u8; 16]> {
184836×
        self.repo.id()
185836×
    }
186835×
    fn refs(&self) -> Result<RefsResponse> {
187835×
        self.repo.remote_refs()
188835×
    }
189313×
    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>> {
190313×
        self.repo.negotiate(have, want)
191313×
    }
192316×
    fn fetch(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle> {
193316×
        self.repo.bundle_raw_for(changes, max_blob)
194316×
    }
1953×
    fn objects(&self, refs: &[ObjectRef]) -> Result<Vec<(ObjectType, Vec<u8>)>> {
1963×
        self.repo.objects_for(refs)
1973×
    }
1981×
    fn checkpoint(&self, heads: &[ChangeHash]) -> Result<Bundle> {
1991×
        self.repo.checkpoint_bundle_for(heads)
2001×
    }
2013×
    fn shallow(&self, heads: &[ChangeHash], depth: u32) -> Result<Vec<ChangeHash>> {
2023×
        self.repo.shallow_changes(heads, depth)
2033×
    }
204248×
    fn push(
205248×
        &self,
206248×
        bundle: Bundle,
207248×
        updates: Vec<RefUpdate>,
208248×
        policy: Option<SignedPolicy>,
209248×
        tags: Vec<TagRef>,
210248×
    ) -> Result<Vec<RefResult>> {
211248×
        if self.repo.is_unbound() {
21211×
            match self.claimed_id {
21311×
                Some(id) => self.repo.bind_id(id)?,
214
                None => {
2150×
                    return Err(RepoError::Msg(
2160×
                        "the remote is unbound and this client did not say which repository it is"
2170×
                            .into(),
2180×
                    ))
219
                }
220
            }
221237×
        }
222248×
        self.repo.receive_push(bundle, updates, policy, tags)
223248×
    }
224
}
225
226
// ---------------------------------------------------------------- http
227
228
/// How to reach a remote: credentials and trust.
229
#[derive(Clone, Debug, Default)]
230
pub struct RemoteOptions {
231
    /// Bearer token for pushes.
232
    pub token: Option<String>,
233
    /// PEM file with extra root certificates to trust for `https://` remotes
234
    /// (self-hosted servers); the system web roots are always trusted.
235
    pub ca: Option<PathBuf>,
236
    /// The local repository's id, sent as `Oxy-Repo-Id` so an unbound
237
    /// hosted repository can adopt it on the first push (spec §16.1).
238
    pub repo_id: Option<[u8; 16]>,
239
    /// Where a remote kept on a store (`dir://`, `s3://`, …) keeps its
240
    /// mirror and what it has seen: the remote's directory under
241
    /// `.oxy/remotes`. None (a clone, before there is a repository) means
242
    /// a temporary directory.
243
    pub cache_dir: Option<PathBuf>,
244
    /// Other tokens for the same hub, tried in turn when `token` is
245
    /// refused (401, 403 or 404): those of enclosing checkouts, when
246
    /// `token` came from the credential store rather than from this
247
    /// remote's own configuration (oxygen#112).
248
    pub fallback_tokens: Vec<String>,
249
}
250
251
/// Opens a store for a URL it understands, or passes (`None`). Stores that
252
/// need more than the standard library (the cloud ones) are added by the
253
/// program that links them, with [`register_store_opener`].
254
pub type StoreOpener =
255
    fn(&str, &RemoteOptions) -> Option<crate::strata::Result<Box<dyn crate::strata::Store>>>;
256
257
static STORE_OPENERS: Mutex<Vec<StoreOpener>> = Mutex::new(Vec::new());
258
259154×
pub fn register_store_opener(opener: StoreOpener) {
260154×
    STORE_OPENERS
261154×
        .lock()
262154×
        .unwrap_or_else(|p| p.into_inner())
263154×
        .push(opener);
264154×
}
265
266
/// A store for `url`, from `dir://` or any registered opener.
2671123×
pub fn open_store(
2681123×
    url: &str,
2691123×
    options: &RemoteOptions,
2701123×
) -> Option<crate::strata::Result<Box<dyn crate::strata::Store>>> {
2711123×
    if let Some(path) = url.strip_prefix("dir://") {
272
        return Some(
27359×
            crate::strata::LocalStore::open(path)
27459×
                .map(|s| Box::new(s) as Box<dyn crate::strata::Store>),
275
        );
2761064×
    }
2771064×
    let openers = STORE_OPENERS.lock().unwrap_or_else(|p| p.into_inner());
2781064×
    openers.iter().find_map(|open| open(url, options))
2791123×
}
280
281
/// The request header carrying the client's repository id.
282
pub const REPO_ID_HEADER: &str = "Oxy-Repo-Id";
283
284
/// The id of a hosted repository that has not adopted one yet.
285
pub const UNBOUND_ID: [u8; 16] = [0; 16];
286
287
/// A remote served by `oxy-server` at `<base>/oxy/v1/...`.
288
pub struct HttpRemote {
289
    base: String,
290
    /// The token in use, and those still to try when it is refused.
291
    token: Mutex<Option<String>>,
292
    fallbacks: Mutex<Vec<String>>,
293
    repo_id: Option<[u8; 16]>,
294
    agent: ureq::Agent,
295
}
296
297
/// A TLS client configuration trusting the web roots plus `ca`, if given:
298
/// the trust every oxy connection uses, shared with tools that call hub APIs.
299147×
pub fn tls_client_config(ca: Option<&Path>) -> Result<Arc<rustls::ClientConfig>> {
300147×
    let mut roots = rustls::RootCertStore::empty();
301147×
    roots.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
302147×
    if let Some(path) = ca {
30335×
        let pem = std::fs::read(path)
30435×
            .map_err(|e| RepoError::Msg(format!("reading CA file {}: {e}", path.display())))?;
30535×
        let certs = rustls_pemfile::certs(&mut pem.as_slice())
30635×
            .collect::<std::result::Result<Vec<_>, _>>()
30735×
            .map_err(|e| RepoError::Msg(format!("parsing CA file {}: {e}", path.display())))?;
30835×
        if certs.is_empty() {
3090×
            return Err(RepoError::Msg(format!(
3100×
                "CA file {} holds no certificates",
3110×
                path.display()
3120×
            )));
31335×
        }
31435×
        for c in certs {
31535×
            roots
31635×
                .add(c)
31735×
                .map_err(|e| RepoError::Msg(format!("CA file {}: {e}", path.display())))?;
318
        }
319112×
    }
320147×
    let cfg = rustls::ClientConfig::builder_with_provider(Arc::new(
321147×
        rustls::crypto::ring::default_provider(),
322
    ))
323147×
    .with_safe_default_protocol_versions()
324147×
    .map_err(|e| RepoError::Msg(e.to_string()))?
325147×
    .with_root_certificates(roots)
326147×
    .with_no_client_auth();
327147×
    Ok(Arc::new(cfg))
328147×
}
329
330
impl HttpRemote {
331143×
    pub fn new(base: &str, options: &RemoteOptions) -> Result<HttpRemote> {
332143×
        refuse_plain_token(base, options)?;
333143×
        let agent = ureq::AgentBuilder::new()
334143×
            .tls_config(tls_client_config(options.ca.as_deref())?)
335143×
            .build();
336143×
        Ok(HttpRemote {
337143×
            base: base.trim_end_matches('/').to_string(),
338143×
            token: Mutex::new(options.token.clone()),
339143×
            fallbacks: Mutex::new(options.fallback_tokens.iter().rev().cloned().collect()),
340143×
            repo_id: options.repo_id,
341143×
            agent,
342143×
        })
343143×
    }
344
345
    /// A request, tried again with the next fallback token while the
346
    /// hub refuses the one in use (oxygen#112).
347356×
    fn call(&self, post: bool, endpoint: &str, body: Option<&[u8]>) -> Result<Vec<u8>> {
348
        loop {
349357×
            let token = self.token.lock().unwrap_or_else(|p| p.into_inner()).clone();
350357×
            match self.call_once(post, endpoint, body, token.as_deref()) {
35111×
                Err(RepoError::Msg(m)) if token.is_some() && refused(&m) => {
3523×
                    let next = self
3533×
                        .fallbacks
3543×
                        .lock()
3553×
                        .unwrap_or_else(|p| p.into_inner())
3563×
                        .pop();
3573×
                    let Some(next) = next else {
3582×
                        return Err(RepoError::Msg(m));
359
                    };
3601×
                    crate::credentials::warn_once(&format!(
3611×
                        "{}: the stored token was refused; using the one an enclosing checkout has for this hub",
3621×
                        self.base
3631×
                    ));
3641×
                    *self.token.lock().unwrap_or_else(|p| p.into_inner()) = Some(next);
365
                }
366354×
                other => return other,
367
            }
368
        }
369356×
    }
370
371357×
    fn call_once(
372357×
        &self,
373357×
        post: bool,
374357×
        endpoint: &str,
375357×
        body: Option<&[u8]>,
376357×
        token: Option<&str>,
377357×
    ) -> Result<Vec<u8>> {
378357×
        let url = format!("{}/oxy/v1/{endpoint}", self.base);
379357×
        let mut req = if post {
380122×
            self.agent.post(&url)
381
        } else {
382235×
            self.agent.get(&url)
383
        };
384357×
        req = req.set("Accept", CONTENT_TYPE);
385357×
        if let Some(t) = token {
38622×
            req = req.set("Authorization", &format!("Bearer {t}"));
387335×
        }
388357×
        if let Some(id) = &self.repo_id {
389332×
            req = req.set(REPO_ID_HEADER, &hex::encode(id));
390332×
        }
391357×
        let resp = match body {
392122×
            Some(b) => req.set("Content-Type", CONTENT_TYPE).send_bytes(b),
393235×
            None => req.call(),
394
        };
39511×
        match resp {
396346×
            Ok(r) => {
397346×
                let mut out = Vec::new();
398346×
                r.into_reader()
399346×
                    .read_to_end(&mut out)
400346×
                    .map_err(|e| RepoError::Msg(format!("{url}: {e}")))?;
401346×
                Ok(out)
402
            }
4038×
            Err(ureq::Error::Status(code, r)) => {
4048×
                let text = r.into_string().unwrap_or_default();
4058×
                let msg = format!("{url}: HTTP {code}: {}", text.trim());
4068×
                Err(RepoError::Msg(with_hint(msg)))
407
            }
408
            // The transport error already names the URL.
4093×
            Err(e) => Err(RepoError::Msg(with_hint(e.to_string()))),
410
        }
411357×
    }
412
}
413
414
/// Whether a failed request's message says the hub refused its token:
415
/// HTTP 401, 403, or 404 (a hub hides a private repository from a token
416
/// that may not read it).
4174×
fn refused(message: &str) -> bool {
4184×
    [": HTTP 401", ": HTTP 403", ": HTTP 404"]
4194×
        .iter()
4206×
        .any(|c| message.contains(c))
4214×
}
422
423
/// Refuse to send a token where anyone on the way could read it: plain
424
/// `http` to another machine (see [`crate::credentials::token_allowed`]).
425143×
fn refuse_plain_token(url: &str, options: &RemoteOptions) -> Result<()> {
426143×
    if options.token.is_some() && !crate::credentials::token_allowed(url) {
4270×
        return Err(RepoError::Msg(format!(
4280×
            "{url}: refusing to send a token over plain http; use https"
4290×
        )));
430143×
    }
431143×
    Ok(())
432143×
}
433
434
/// What to do about a failed request, recognised from its message.
43519×
pub fn hint_for(message: &str) -> Option<&'static str> {
43619×
    let m = message.to_ascii_lowercase();
43719×
    let hint = if m.contains("http 401") {
4386×
        "the hub needs you signed in, or did not accept the token: run `oxy auth login HUB` (pushing needs repo:write)"
43913×
    } else if m.contains("http 403") {
4401×
        "the token or account lacks permission: pushing needs a token with repo:write and write access to the repository"
44112×
    } else if m.contains("http 404") {
4424×
        "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`)"
4438×
    } else if m.contains("unknownissuer")
4446×
        || m.contains("unknown issuer")
4456×
        || m.contains("certificate")
4465×
        || m.contains("invalidcertificate")
447
    {
4483×
        "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`"
4495×
    } else if m.contains("connection refused") {
4503×
        "nothing is listening there: check the URL and port, and that the server is running"
4512×
    } else if m.contains("dns")
4521×
        || m.contains("failed to lookup")
4531×
        || m.contains("name or service not known")
454
    {
4551×
        "the host name does not resolve: check the URL"
4561×
    } else if m.contains("timed out") || m.contains("timeout") {
4570×
        "the server did not answer in time: check the network, or try again"
458
    } else {
4591×
        return None;
460
    };
46118×
    Some(hint)
46219×
}
463
464
/// `message`, followed by what to do about it when that is known.
46511×
pub fn with_hint(message: String) -> String {
46611×
    match hint_for(&message) {
46711×
        Some(h) => format!("{message}\n  hint: {h}"),
4680×
        None => message,
469
    }
47011×
}
471
472
/// A JSON request to a hub API (`GET`, `POST` or `DELETE`), with the same
473
/// TLS trust and token rules as remotes. Returns the status and body, an
474
/// error status included; only transport failures are errors.
4750×
pub fn http_json(
4760×
    method: &str,
4770×
    url: &str,
4780×
    body: Option<&str>,
4790×
    options: &RemoteOptions,
4800×
) -> Result<(u16, String)> {
4810×
    refuse_plain_token(url, options)?;
4820×
    let agent = ureq::AgentBuilder::new()
4830×
        .tls_config(tls_client_config(options.ca.as_deref())?)
4840×
        .timeout(std::time::Duration::from_secs(60))
4850×
        .build();
4860×
    let mut req = agent.request(method, url).set("Accept", "application/json");
4870×
    if let Some(t) = &options.token {
4880×
        req = req.set("Authorization", &format!("Bearer {t}"));
4890×
    }
4900×
    let resp = match body {
4910×
        Some(b) => req.set("Content-Type", "application/json").send_string(b),
4920×
        None => req.call(),
493
    };
4940×
    match resp {
4950×
        Ok(r) => {
4960×
            let code = r.status();
4970×
            Ok((code, r.into_string().unwrap_or_default()))
498
        }
4990×
        Err(ureq::Error::Status(code, r)) => Ok((code, r.into_string().unwrap_or_default())),
5000×
        Err(e) => Err(RepoError::Msg(with_hint(e.to_string()))),
501
    }
5020×
}
503
504
/// An authenticated GET of any URL on a host, with the same TLS trust and
505
/// token as remotes use (for hub APIs: proposals). Returns the body.
5060×
pub fn http_get(url: &str, options: &RemoteOptions) -> Result<Vec<u8>> {
5070×
    refuse_plain_token(url, options)?;
5080×
    let agent = ureq::AgentBuilder::new()
5090×
        .tls_config(tls_client_config(options.ca.as_deref())?)
5100×
        .build();
5110×
    let mut req = agent.get(url).set("Accept", "application/json");
5120×
    if let Some(t) = &options.token {
5130×
        req = req.set("Authorization", &format!("Bearer {t}"));
5140×
    }
5150×
    match req.call() {
5160×
        Ok(r) => {
5170×
            let mut out = Vec::new();
5180×
            r.into_reader()
5190×
                .read_to_end(&mut out)
5200×
                .map_err(|e| RepoError::Msg(format!("{url}: {e}")))?;
5210×
            Ok(out)
522
        }
5230×
        Err(ureq::Error::Status(code, r)) => {
5240×
            let text = r.into_string().unwrap_or_default();
5250×
            Err(RepoError::Msg(with_hint(format!(
5260×
                "{url}: HTTP {code}: {}",
5270×
                text.trim()
5280×
            ))))
529
        }
530
        // The transport error already names the URL.
5310×
        Err(e) => Err(RepoError::Msg(with_hint(e.to_string()))),
532
    }
5330×
}
534
535
impl Remote for HttpRemote {
536122×
    fn repo_id(&self) -> Result<[u8; 16]> {
537122×
        wire::decode_capabilities(&self.call(false, "capabilities", None)?)
538122×
    }
539113×
    fn refs(&self) -> Result<RefsResponse> {
540113×
        Ok(wire::decode_refs(&self.call(false, "refs", None)?)?.1)
541113×
    }
54244×
    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>> {
54344×
        wire::decode_hashes(&self.call(
544
            true,
54544×
            "negotiate",
54644×
            Some(&wire::encode_negotiate(have, want)),
5470×
        )?)
54844×
    }
54944×
    fn fetch(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle> {
55044×
        wire::decode_bundle(&self.call(
551
            true,
55244×
            "fetch",
55344×
            Some(&wire::encode_fetch(changes, max_blob)),
5540×
        )?)
55544×
    }
5561×
    fn objects(&self, refs: &[ObjectRef]) -> Result<Vec<(ObjectType, Vec<u8>)>> {
557
        Ok(
5581×
            wire::decode_bundle(&self.call(true, "objects", Some(&wire::encode_hashes(refs)))?)?
559
                .objects,
560
        )
5611×
    }
5620×
    fn checkpoint(&self, heads: &[ChangeHash]) -> Result<Bundle> {
5630×
        wire::decode_bundle(&self.call(true, "checkpoint", Some(&wire::encode_hashes(heads)))?)
5640×
    }
5650×
    fn shallow(&self, heads: &[ChangeHash], depth: u32) -> Result<Vec<ChangeHash>> {
5660×
        wire::decode_hashes(&self.call(
567
            true,
5680×
            "shallow",
5690×
            Some(&wire::encode_shallow(heads, depth)),
5700×
        )?)
5710×
    }
57232×
    fn push(
57332×
        &self,
57432×
        bundle: Bundle,
57532×
        updates: Vec<RefUpdate>,
57632×
        policy: Option<SignedPolicy>,
57732×
        tags: Vec<TagRef>,
57832×
    ) -> Result<Vec<RefResult>> {
57932×
        wire::decode_push_response(&self.call(
580
            true,
58132×
            "push",
58232×
            Some(&wire::encode_push(
58332×
                &bundle,
58432×
                &updates,
58532×
                policy.as_ref(),
58632×
                &tags,
58732×
            )),
5884×
        )?)
58932×
    }
590
}
591
592
/// A remote's URL as it is stored: an `http(s)://` URL as given, a local
593
/// path made absolute (with its `file://` kept), so it still names the same
594
/// repository from inside the clone (oxygen#57).
595258×
pub fn stored_url(url: &str) -> String {
596
    // Only a path is made absolute: a plain one, or one behind file:// or
597
    // dir://. Every other URL (a hub, a bucket) is kept as written.
598258×
    let (prefix, path) = match (url.strip_prefix("file://"), url.strip_prefix("dir://")) {
5991×
        (Some(p), _) => ("file://", p),
60011×
        (_, Some(p)) => ("dir://", p),
601246×
        _ if url.contains("://") => return url.to_string(),
602223×
        _ => ("", url),
603
    };
604235×
    match std::path::absolute(path) {
605235×
        Ok(abs) => format!("{prefix}{}", abs.display()),
6060×
        Err(_) => url.to_string(),
607
    }
608258×
}
609
610
/// Open a remote by URL: a local path (optionally `file://`), `http(s)://`
611
/// (a hub), or a store (`dir://`, and whatever openers are registered).
6121254×
pub fn open_remote(url: &str, options: &RemoteOptions) -> Result<Box<dyn Remote>> {
6131254×
    if url.starts_with("http://") || url.starts_with("https://") {
614143×
        return Ok(Box::new(HttpRemote::new(url, options)?));
6151111×
    }
6161111×
    if let Some(store) = open_store(url, options) {
61748×
        return crate::repo::strata::open_store_remote(store?, options);
6181063×
    }
6191063×
    let path = url.strip_prefix("file://").unwrap_or(url);
6201063×
    if path.contains("://") {
6210×
        return Err(RepoError::Msg(format!("unsupported remote URL {url:?}")));
6221063×
    }
6231063×
    let mut local = LocalRemote::open(Path::new(path))?;
6241063×
    local.claimed_id = options.repo_id;
6251063×
    Ok(Box::new(local))
6261254×
}
627
628
// ---------------------------------------------------------------- wire
629
630
/// CBOR encodings of the protocol messages (spec §16.1).
631
pub mod wire {
632
    use super::*;
633
6340×
    fn schema(msg: &str) -> RepoError {
6350×
        RepoError::Object(ObjectError::Schema(msg.into()))
6360×
    }
637145×
    fn sigs_value(sigs: &[Signature]) -> Value {
638145×
        Value::Array(sigs.iter().map(|s| s.to_value()).collect())
639145×
    }
640140×
    fn sigs_from(v: &Value) -> Result<Vec<Signature>> {
641140×
        Ok(o::array(v, "signatures")?
642140×
            .iter()
643140×
            .map(Signature::from_value)
644140×
            .collect::<std::result::Result<Vec<_>, _>>()?)
645140×
    }
646336×
    fn hashes_value(hs: &[ObjectRef]) -> Value {
647488×
        Value::Array(hs.iter().map(|h| Value::Bytes(h.0.to_vec())).collect())
648336×
    }
649395×
    fn hashes_from(v: &Value) -> Result<Vec<ObjectRef>> {
650395×
        Ok(o::array(v, "hashes")?
651395×
            .iter()
652548×
            .map(|h| o::oref(h, "hash"))
653395×
            .collect::<std::result::Result<Vec<_>, _>>()?)
654395×
    }
65524×
    fn attest_value(sa: &SignedAttest) -> Value {
65624×
        Value::Array(vec![Value::Bytes(sa.attest.encode()), sigs_value(&sa.sigs)])
65724×
    }
65838×
    fn attest_from(v: &Value) -> Result<SignedAttest> {
65938×
        let a = o::array(v, "signed attest")?;
66038×
        if a.len() != 2 {
6610×
            return Err(schema("signed attest: expected [attest, sigs]"));
66238×
        }
663
        Ok(SignedAttest {
66438×
            attest: Attest::decode(o::bytes(&a[0], "attest")?)?,
66538×
            sigs: sigs_from(&a[1])?,
666
        })
66738×
    }
668185×
    fn opt_attest_value(sa: &Option<SignedAttest>) -> Value {
669185×
        sa.as_ref().map(attest_value).unwrap_or(Value::Null)
670185×
    }
671240×
    fn opt_attest_from(v: &Value) -> Result<Option<SignedAttest>> {
672240×
        if v.is_null() {
673213×
            Ok(None)
674
        } else {
67527×
            attest_from(v).map(Some)
676
        }
677240×
    }
678
    /// `[policy bytes, [signature]]`: also the format of a policy proposal
679
    /// file exchanged between root holders (spec §14.5).
68010×
    pub fn policy_value(sp: &SignedPolicy) -> Value {
68110×
        Value::Array(vec![Value::Bytes(sp.policy.encode()), sigs_value(&sp.sigs)])
68210×
    }
6838×
    pub fn policy_from(v: &Value) -> Result<SignedPolicy> {
6848×
        let a = o::array(v, "signed policy")?;
6858×
        if a.len() != 2 {
6860×
            return Err(schema("signed policy: expected [policy, sigs]"));
6878×
        }
688
        Ok(SignedPolicy {
6898×
            policy: Policy::decode(o::bytes(&a[0], "policy")?)?,
6908×
            sigs: sigs_from(&a[1])?,
691
        })
6928×
    }
693174×
    fn opt_policy_value(sp: Option<&SignedPolicy>) -> Value {
694174×
        sp.map(policy_value).unwrap_or(Value::Null)
695174×
    }
696213×
    fn opt_policy_from(v: &Value) -> Result<Option<SignedPolicy>> {
697213×
        if v.is_null() {
698210×
            Ok(None)
699
        } else {
7003×
            policy_from(v).map(Some)
701
        }
702213×
    }
703871×
    fn map<'a>(v: &'a Value, what: &str) -> Result<&'a o::Map> {
704871×
        Ok(o::map(v, what)?)
705871×
    }
7062450×
    fn field<'a>(m: &'a o::Map, k: u64, what: &str) -> Result<&'a Value> {
7072450×
        Ok(o::field(m, k, what)?)
7082450×
    }
709
710
    /// A bundle's value. Each change's text blobs and their chunks (spec
711
    /// §6.1) are added to its objects when they are not already there, so a
712
    /// bundle can always be read on its own.
713113×
    pub fn bundle_value(b: &Bundle) -> Value {
714113×
        let mut changes = Vec::with_capacity(b.raw_changes.len() + b.changes.len());
715
        // Changes as stored are already their encoding, their texts among
716
        // the objects.
717113×
        for (bytes, s) in &b.raw_changes {
7180×
            changes.push(Value::Array(vec![
7190×
                Value::Bytes(bytes.clone()),
7200×
                sigs_value(s),
7210×
            ]));
7220×
        }
723113×
        let mut texts: Vec<(ObjectType, Vec<u8>)> = Vec::new();
724113×
        let mut listed: Option<std::collections::HashSet<ObjectRef>> = None;
725113×
        for (c, s) in &b.changes {
726111×
            let (bytes, stored) = c.encode_parts();
727111×
            for t in stored {
7280×
                let listed = listed.get_or_insert_with(|| {
7290×
                    b.objects
7300×
                        .iter()
7310×
                        .map(|(ty, body)| crate::hash::hash_object(*ty, body))
7320×
                        .collect()
7330×
                });
7340×
                if !listed.insert(t.blob.hash()) {
7350×
                    continue;
7360×
                }
7370×
                for chunk in crate::chunk::chunks(&t.content) {
7380×
                    if listed.insert(crate::hash::hash_object(ObjectType::Chunk, chunk)) {
7390×
                        texts.push((ObjectType::Chunk, chunk.to_vec()));
7400×
                    }
741
                }
7420×
                texts.push((ObjectType::Blob, t.blob.encode()));
743
            }
744111×
            changes.push(Value::Array(vec![Value::Bytes(bytes), sigs_value(s)]));
745
        }
746113×
        let mut v = Value::map([
747113×
            (0, Value::Array(changes)),
748113×
            (
749113×
                1,
750113×
                Value::Array(b.attests.iter().map(attest_value).collect()),
751113×
            ),
752
            (
753
                2,
754
                Value::Array(
755113×
                    b.objects
756113×
                        .iter()
757113×
                        .chain(&texts)
758113×
                        .map(|(t, body)| {
75986×
                            Value::Array(vec![
76086×
                                Value::Text(t.name().into()),
76186×
                                Value::Bytes(body.clone()),
76286×
                            ])
76386×
                        })
764113×
                        .collect(),
765
                ),
766
            ),
767113×
            (
768113×
                3,
769113×
                Value::Array(b.policies.iter().map(policy_value).collect()),
770113×
            ),
771
        ]);
772
        // Only when there are any; older peers ignore the field.
773113×
        if !b.certificates.is_empty() {
7741×
            if let Value::Map(m) = &mut v {
7751×
                m.insert(
776
                    4,
777
                    Value::Array(
7781×
                        b.certificates
7791×
                            .iter()
7801×
                            .map(|c| Value::Bytes(c.encode()))
7811×
                            .collect(),
782
                    ),
783
                );
7840×
            }
785112×
        }
786113×
        v
787113×
    }
788
789103×
    pub fn bundle_from_value(v: &Value) -> Result<Bundle> {
790103×
        let m = map(v, "bundle")?;
791103×
        let mut b = Bundle::default();
792103×
        for ob in o::array(field(m, 2, "objects")?, "objects")? {
79375×
            let pair = o::array(ob, "object")?;
79475×
            if pair.len() != 2 {
7950×
                return Err(schema("object: expected [type, body]"));
79675×
            }
79775×
            let ty = ObjectType::from_name(o::text(&pair[0], "object type")?)
79875×
                .ok_or_else(|| schema("unknown object type"))?;
79975×
            b.objects
80075×
                .push((ty, o::bytes(&pair[1], "object body")?.to_vec()));
801
        }
802
        // Stored inserts are read from the bundle's own objects.
803103×
        let mut by_ref: Option<std::collections::HashMap<ObjectRef, usize>> = None;
804103×
        for c in o::array(field(m, 0, "changes")?, "changes")? {
80594×
            let pair = o::array(c, "change entry")?;
80694×
            if pair.len() != 2 {
8070×
                return Err(schema("change entry: expected [change, sigs]"));
80894×
            }
80994×
            let encoded = o::bytes(&pair[0], "change")?;
81094×
            b.hashes
81194×
                .push(crate::hash::hash_object(ObjectType::Change, encoded));
81294×
            let change = Change::decode_with(encoded, &mut |text, count, form| {
8130×
                let by_ref = by_ref.get_or_insert_with(|| {
8140×
                    b.objects
8150×
                        .iter()
8160×
                        .enumerate()
8170×
                        .map(|(i, (ty, body))| (crate::hash::hash_object(*ty, body), i))
8180×
                        .collect()
8190×
                });
8200×
                let get =
8210×
                    |r: &ObjectRef, want: ObjectType| match by_ref.get(r).map(|i| &b.objects[*i]) {
8220×
                        Some((ty, body)) if *ty == want => Ok(body),
8230×
                        _ => Err(ObjectError::Schema(format!(
8240×
                            "the bundle lacks {} {}",
8250×
                            want.name(),
8260×
                            r.short(12)
8270×
                        ))),
8280×
                    };
8290×
                let blob = crate::objects::Blob::decode(get(text, ObjectType::Blob)?)?;
8300×
                let mut content = Vec::with_capacity(blob.length as usize);
8310×
                for chunk in &blob.chunks {
8320×
                    content.extend_from_slice(get(chunk, ObjectType::Chunk)?);
833
                }
8340×
                crate::objects::text_lines(&content, count, form)
8350×
            })?;
83694×
            b.changes.push((change, sigs_from(&pair[1])?));
837
        }
838103×
        for a in o::array(field(m, 1, "attests")?, "attests")? {
83911×
            b.attests.push(attest_from(a)?);
840
        }
841103×
        if let Some(ps) = m.get(&3) {
842103×
            for p in o::array(ps, "policies")? {
8433×
                b.policies.push(policy_from(p)?);
844
            }
8450×
        }
846103×
        if let Some(cs) = m.get(&4) {
8471×
            for c in o::array(cs, "certificates")? {
8481×
                b.certificates.push(
8491×
                    crate::cert::SignedCertificate::decode(o::bytes(c, "certificate")?)
8501×
                        .map_err(|e| schema(&format!("certificate: {e}")))?,
851
                );
852
            }
853102×
        }
854103×
        Ok(b)
855103×
    }
856
85780×
    pub fn encode_bundle(b: &Bundle) -> Vec<u8> {
85880×
        cbor::encode(&bundle_value(b))
85980×
    }
86074×
    pub fn decode_bundle(bytes: &[u8]) -> Result<Bundle> {
86174×
        bundle_from_value(&cbor::decode(bytes).map_err(ObjectError::from)?)
86274×
    }
863
864117×
    pub fn encode_capabilities(id: [u8; 16]) -> Vec<u8> {
865117×
        cbor::encode(&Value::map([
866117×
            (0, Value::Array(vec![Value::Uint(PROTOCOL_VERSION)])),
867117×
            (1, Value::Bytes(id.to_vec())),
868117×
            (2, Value::Array(vec![Value::Text("seal".into())])),
869117×
        ]))
870117×
    }
871117×
    pub fn decode_capabilities(bytes: &[u8]) -> Result<[u8; 16]> {
872117×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
873117×
        let m = map(&v, "capabilities")?;
874117×
        let versions = o::array(field(m, 0, "versions")?, "versions")?;
875117×
        if !versions
876117×
            .iter()
877117×
            .any(|x| x.as_uint() == Some(PROTOCOL_VERSION))
878
        {
8790×
            return Err(RepoError::Msg(
8800×
                "remote does not speak protocol version 1".into(),
8810×
            ));
882117×
        }
883117×
        o::bytes(field(m, 1, "repo id")?, "repo id")?
884117×
            .try_into()
885117×
            .map_err(|_| schema("repo id: expected 16 bytes"))
886117×
    }
887
888174×
    fn tags_value(tags: &[TagRef]) -> Value {
889
        Value::Array(
890174×
            tags.iter()
891174×
                .map(|t| {
8927×
                    Value::map([
8937×
                        (0, Value::Text(t.name.clone())),
8947×
                        (1, Value::Bytes(t.state.0.to_vec())),
8957×
                        (2, hashes_value(&t.heads)),
8967×
                        (3, opt_attest_value(&t.attestation)),
8977×
                    ])
8987×
                })
899174×
                .collect(),
900
        )
901174×
    }
902213×
    fn tags_from(v: &Value) -> Result<Vec<TagRef>> {
903213×
        let mut out = Vec::new();
904213×
        for t in o::array(v, "tags")? {
90523×
            let tm = map(t, "tag")?;
90623×
            out.push(TagRef {
90723×
                name: o::text(field(tm, 0, "name")?, "name")?.to_owned(),
90823×
                state: o::oref(field(tm, 1, "state")?, "state")?,
90923×
                heads: hashes_from(field(tm, 2, "heads")?)?,
91023×
                attestation: opt_attest_from(field(tm, 3, "attestation")?)?,
911
            });
912
        }
913213×
        Ok(out)
914213×
    }
915
916141×
    pub fn encode_refs(id: [u8; 16], resp: &RefsResponse) -> Vec<u8> {
917141×
        cbor::encode(&Value::map([
918
            (
919
                0,
920
                Value::Array(
921141×
                    resp.refs
922141×
                        .iter()
923145×
                        .map(|r| {
924145×
                            let mut v = Value::map([
925145×
                                (0, Value::Text(r.name.clone())),
926145×
                                (1, Value::Bytes(r.state.0.to_vec())),
927145×
                                (2, hashes_value(&r.heads)),
928145×
                                (3, opt_attest_value(&r.attestation)),
929145×
                                (4, r.last_serial.map(Value::Uint).unwrap_or(Value::Null)),
930145×
                            ]);
931
                            // Only when there is one; older peers ignore it.
932145×
                            if r.accepted.is_some() {
9330×
                                if let Value::Map(m) = &mut v {
9340×
                                    m.insert(5, opt_attest_value(&r.accepted));
9350×
                                }
936145×
                            }
937145×
                            v
938145×
                        })
939141×
                        .collect(),
940
                ),
941
            ),
942141×
            (1, Value::Bytes(id.to_vec())),
943141×
            (2, opt_policy_value(resp.policy.as_ref())),
944141×
            (3, tags_value(&resp.tags)),
945141×
            (
946141×
                4,
947141×
                Value::Array(resp.history.iter().map(policy_value).collect()),
948141×
            ),
949141×
            (
950141×
                5,
951141×
                resp.default_bookmark
952141×
                    .clone()
953141×
                    .map(Value::Text)
954141×
                    .unwrap_or(Value::Null),
955141×
            ),
956
        ]))
957141×
    }
958184×
    pub fn decode_refs(bytes: &[u8]) -> Result<([u8; 16], RefsResponse)> {
959184×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
960184×
        let m = map(&v, "refs")?;
961184×
        let id: [u8; 16] = o::bytes(field(m, 1, "repo id")?, "repo id")?
962184×
            .try_into()
963184×
            .map_err(|_| schema("repo id: expected 16 bytes"))?;
964184×
        let mut refs = Vec::new();
965188×
        for r in o::array(field(m, 0, "refs")?, "refs")? {
966188×
            let rm = map(r, "ref")?;
967188×
            let ls = field(rm, 4, "last serial")?;
968188×
            refs.push(RemoteRef {
969188×
                name: o::text(field(rm, 0, "name")?, "name")?.to_owned(),
970188×
                state: o::oref(field(rm, 1, "state")?, "state")?,
971188×
                heads: hashes_from(field(rm, 2, "heads")?)?,
972188×
                attestation: opt_attest_from(field(rm, 3, "attestation")?)?,
973188×
                last_serial: if ls.is_null() {
974172×
                    None
975
                } else {
97616×
                    Some(o::uint(ls, "last serial")?)
977
                },
978188×
                accepted: match rm.get(&5) {
9790×
                    Some(a) => opt_attest_from(a)?,
980188×
                    None => None,
981
                },
982
            });
983
        }
984184×
        let policy = match m.get(&2) {
985184×
            Some(p) => opt_policy_from(p)?,
9860×
            None => None,
987
        };
988184×
        let tags = match m.get(&3) {
989184×
            Some(t) => tags_from(t)?,
9900×
            None => Vec::new(),
991
        };
992184×
        let mut history = Vec::new();
993184×
        if let Some(h) = m.get(&4) {
994184×
            for p in o::array(h, "history")? {
9951×
                history.push(policy_from(p)?);
996
            }
9970×
        }
998184×
        let default_bookmark = match m.get(&5) {
999115×
            Some(Value::Text(t)) => Some(t.clone()),
100069×
            _ => None,
1001
        };
1002184×
        Ok((
1003184×
            id,
1004184×
            RefsResponse {
1005184×
                refs,
1006184×
                policy,
1007184×
                tags,
1008184×
                history,
1009184×
                default_bookmark,
1010184×
            },
1011184×
        ))
1012184×
    }
1013
101445×
    pub fn encode_negotiate(have: &[ChangeHash], want: &[ChangeHash]) -> Vec<u8> {
101545×
        cbor::encode(&Value::map([
101645×
            (0, hashes_value(have)),
101745×
            (1, hashes_value(want)),
101845×
        ]))
101945×
    }
102045×
    pub fn decode_negotiate(bytes: &[u8]) -> Result<(Vec<ChangeHash>, Vec<ChangeHash>)> {
102145×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
102245×
        let m = map(&v, "negotiate")?;
1023
        Ok((
102445×
            hashes_from(field(m, 0, "have")?)?,
102545×
            hashes_from(field(m, 1, "want")?)?,
1026
        ))
102745×
    }
1028
1029
    /// `fetch` request: `{0: [changes], 1: max blob size or null}`.
103046×
    pub fn encode_fetch(hs: &[ChangeHash], max_blob: Option<u64>) -> Vec<u8> {
103146×
        cbor::encode(&Value::map([
103246×
            (0, hashes_value(hs)),
103346×
            (1, max_blob.map(Value::Uint).unwrap_or(Value::Null)),
103446×
        ]))
103546×
    }
103647×
    pub fn decode_fetch(bytes: &[u8]) -> Result<(Vec<ChangeHash>, Option<u64>)> {
103747×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
103847×
        let m = map(&v, "fetch")?;
103947×
        let max = match m.get(&1) {
104046×
            Some(x) if !x.is_null() => Some(o::uint(x, "max blob")?),
104145×
            _ => None,
1042
        };
104347×
        Ok((hashes_from(field(m, 0, "changes")?)?, max))
104447×
    }
1045
1046
    /// `shallow` request: `{0: [heads], 1: depth}`.
10471×
    pub fn encode_shallow(hs: &[ChangeHash], depth: u32) -> Vec<u8> {
10481×
        cbor::encode(&Value::map([
10491×
            (0, hashes_value(hs)),
10501×
            (1, Value::Uint(depth as u64)),
10511×
        ]))
10521×
    }
10531×
    pub fn decode_shallow(bytes: &[u8]) -> Result<(Vec<ChangeHash>, u32)> {
10541×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
10551×
        let m = map(&v, "shallow")?;
1056
        Ok((
10571×
            hashes_from(field(m, 0, "heads")?)?,
10581×
            o::u32_(field(m, 1, "depth")?, "depth")?,
1059
        ))
10601×
    }
1061
106247×
    pub fn encode_hashes(hs: &[ChangeHash]) -> Vec<u8> {
106347×
        cbor::encode(&Value::map([(0, hashes_value(hs))]))
106447×
    }
106546×
    pub fn decode_hashes(bytes: &[u8]) -> Result<Vec<ChangeHash>> {
106646×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
106746×
        hashes_from(field(map(&v, "hashes")?, 0, "hashes")?)
106846×
    }
1069
107033×
    pub fn encode_push(
107133×
        bundle: &Bundle,
107233×
        updates: &[RefUpdate],
107333×
        policy: Option<&SignedPolicy>,
107433×
        tags: &[TagRef],
107533×
    ) -> Vec<u8> {
107633×
        cbor::encode(&Value::map([
107733×
            (0, bundle_value(bundle)),
107833×
            (2, opt_policy_value(policy)),
107933×
            (3, tags_value(tags)),
1080
            (
1081
                1,
1082
                Value::Array(
108333×
                    updates
108433×
                        .iter()
108533×
                        .map(|u| {
108633×
                            Value::map([
108733×
                                (0, Value::Text(u.name.clone())),
108833×
                                (1, Value::Bytes(u.old_state.0.to_vec())),
108933×
                                (2, Value::Bytes(u.new_state.0.to_vec())),
109033×
                                (3, Value::Bool(u.force)),
109133×
                                (4, opt_attest_value(&u.attestation)),
109233×
                            ])
109333×
                        })
109433×
                        .collect(),
1095
                ),
1096
            ),
1097
        ]))
109833×
    }
1099
    /// A decoded push: bundle, bookmark updates, policy, tags.
1100
    pub type PushParts = (Bundle, Vec<RefUpdate>, Option<SignedPolicy>, Vec<TagRef>);
1101
110229×
    pub fn decode_push(bytes: &[u8]) -> Result<PushParts> {
110329×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
110429×
        let m = map(&v, "push")?;
110529×
        let bundle = bundle_from_value(field(m, 0, "bundle")?)?;
110629×
        let mut updates = Vec::new();
110729×
        for u in o::array(field(m, 1, "updates")?, "updates")? {
110829×
            let um = map(u, "update")?;
110929×
            updates.push(RefUpdate {
111029×
                name: o::text(field(um, 0, "name")?, "name")?.to_owned(),
111129×
                old_state: o::oref(field(um, 1, "old_state")?, "old_state")?,
111229×
                new_state: o::oref(field(um, 2, "new_state")?, "new_state")?,
111329×
                force: field(um, 3, "force")?
111429×
                    .as_bool()
111529×
                    .ok_or_else(|| schema("force: expected bool"))?,
111629×
                attestation: opt_attest_from(field(um, 4, "attestation")?)?,
1117
            });
1118
        }
111929×
        let policy = match m.get(&2) {
112029×
            Some(p) => opt_policy_from(p)?,
11210×
            None => None,
1122
        };
112329×
        let tags = match m.get(&3) {
112429×
            Some(t) => tags_from(t)?,
11250×
            None => Vec::new(),
1126
        };
112729×
        Ok((bundle, updates, policy, tags))
112829×
    }
1129
113029×
    pub fn encode_push_response(results: &[RefResult]) -> Vec<u8> {
113129×
        cbor::encode(&Value::map([(
1132
            0,
1133
            Value::Array(
113429×
                results
113529×
                    .iter()
113630×
                    .map(|r| {
113730×
                        Value::map([
113830×
                            (0, Value::Text(r.name.clone())),
113930×
                            (1, Value::Bool(r.ok)),
114030×
                            (2, Value::Text(r.reason.clone())),
114130×
                        ])
114230×
                    })
114329×
                    .collect(),
1144
            ),
1145
        )]))
114629×
    }
114729×
    pub fn decode_push_response(bytes: &[u8]) -> Result<Vec<RefResult>> {
114829×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
114929×
        let m = map(&v, "push response")?;
115029×
        let mut out = Vec::new();
115130×
        for r in o::array(field(m, 0, "results")?, "results")? {
115230×
            let rm = map(r, "result")?;
115330×
            out.push(RefResult {
115430×
                name: o::text(field(rm, 0, "name")?, "name")?.to_owned(),
115530×
                ok: field(rm, 1, "ok")?
115630×
                    .as_bool()
115730×
                    .ok_or_else(|| schema("ok: expected bool"))?,
115830×
                reason: o::text(field(rm, 2, "reason")?, "reason")?.to_owned(),
1159
            });
1160
        }
116129×
        Ok(out)
116229×
    }
1163
}
1164
1165
#[cfg(test)]
1166
mod tests {
1167
    use super::*;
1168
    use crate::attest::AttestBody;
1169
    use crate::objects::Hlc;
1170
1171
    #[test]
11721×
    fn wire_round_trips() {
11731×
        let c = crate::objects::tests::sample_change();
11741×
        let sk = crate::identity::SecretKey {
11751×
            ed_seed: [1; 32],
11761×
            ml_seed: [2; 32],
11771×
        };
11781×
        let id = sk.identity("a", "b", 0);
11791×
        let key = id.hash();
11801×
        let sig = sk.sign(crate::identity::Domain::Change, &key, &c.hash());
11811×
        let seal = Attest::seal([3; 16], vec![c.hash()]);
11821×
        let seal_sig = sk.sign(crate::identity::Domain::Attest, &key, &seal.hash());
11831×
        let policy = Policy::single_root([3; 16], id.clone(), 99);
11841×
        let psig = sk.sign(crate::identity::Domain::Policy, &key, &policy.hash());
11851×
        let bundle = Bundle {
11861×
            changes: vec![(c.clone(), vec![sig.clone()])],
11871×
            attests: vec![SignedAttest {
11881×
                attest: seal.clone(),
11891×
                sigs: vec![seal_sig],
11901×
            }],
11911×
            objects: vec![
11921×
                (ObjectType::Identity, id.encode()),
11931×
                (ObjectType::Chunk, b"xyz".to_vec()),
11941×
            ],
11951×
            policies: vec![SignedPolicy {
11961×
                policy: policy.clone(),
11971×
                sigs: vec![psig.clone()],
11981×
            }],
11991×
            hashes: vec![],
12001×
            raw_changes: vec![],
12011×
            certificates: vec![],
12021×
        };
12031×
        let d = wire::decode_bundle(&wire::encode_bundle(&bundle)).unwrap();
12041×
        assert_eq!(
1205
            d.hashes,
12061×
            vec![c.hash()],
1207
            "the receiver hashes what it was sent"
1208
        );
12091×
        assert_eq!(d.changes, bundle.changes);
12101×
        assert_eq!(d.attests, bundle.attests);
12111×
        assert_eq!(d.objects, bundle.objects);
12121×
        assert_eq!(d.policies, bundle.policies);
12131×
        assert!(d.certificates.is_empty());
1214
        // Certificates (keyless, oxygenhub#241) travel as field 4, only
1215
        // when there are any: a bundle without them encodes as before.
12161×
        let plain = wire::encode_bundle(&bundle);
12171×
        let mut with = bundle.clone();
12181×
        let issuer_sk = crate::identity::SecretKey::generate();
12191×
        let issuer = issuer_sk.identity("hub", "h@x", 1);
12201×
        with.certificates.push(crate::cert::SignedCertificate::sign(
12211×
            crate::cert::Certificate {
12221×
                key: id.clone(),
12231×
                issuer: issuer.hash(),
12241×
                subject: "agent:a@b".into(),
12251×
                claims: vec![],
12261×
                scope: crate::identity::Scope::CHANGES,
12271×
                not_before: 1,
12281×
                expires: 2,
12291×
                grant: None,
12301×
            },
12311×
            &issuer,
12321×
            &issuer_sk,
1233
        ));
12341×
        let encoded = wire::encode_bundle(&with);
12351×
        assert_ne!(encoded, plain);
12361×
        assert_eq!(
12371×
            wire::decode_bundle(&encoded).unwrap().certificates,
1238
            with.certificates
1239
        );
12401×
        with.certificates.clear();
12411×
        assert_eq!(wire::encode_bundle(&with), plain);
1242
12431×
        let att = Attest {
12441×
            repo: [3; 16],
12451×
            hlc: Hlc::default(),
12461×
            body: AttestBody::Ref {
12471×
                name: "main".into(),
12481×
                old_state: ObjectRef([1; 32]),
12491×
                new_state: ObjectRef([2; 32]),
12501×
                serial: 4,
12511×
            },
12521×
        };
12531×
        let refs = vec![
12541×
            RemoteRef {
12551×
                name: "main".into(),
12561×
                state: ObjectRef([2; 32]),
12571×
                heads: vec![c.hash()],
12581×
                attestation: Some(SignedAttest {
12591×
                    attest: att.clone(),
12601×
                    sigs: vec![sig.clone()],
12611×
                }),
12621×
                last_serial: Some(4),
12631×
                accepted: None,
12641×
            },
12651×
            RemoteRef {
12661×
                name: "dev".into(),
12671×
                state: ObjectRef([5; 32]),
12681×
                heads: vec![],
12691×
                attestation: None,
12701×
                last_serial: None,
12711×
                accepted: None,
12721×
            },
1273
        ];
12741×
        let tag = TagRef {
12751×
            name: "v1".into(),
12761×
            state: ObjectRef([2; 32]),
12771×
            heads: vec![c.hash()],
12781×
            attestation: None,
12791×
        };
12801×
        let resp = RefsResponse {
12811×
            refs,
12821×
            policy: Some(SignedPolicy {
12831×
                policy: policy.clone(),
12841×
                sigs: vec![psig.clone()],
12851×
            }),
12861×
            tags: vec![tag.clone()],
12871×
            history: vec![SignedPolicy {
12881×
                policy: policy.clone(),
12891×
                sigs: vec![],
12901×
            }],
12911×
            default_bookmark: Some("develop".into()),
12921×
        };
12931×
        let (id_back, back) = wire::decode_refs(&wire::encode_refs([3; 16], &resp)).unwrap();
12941×
        assert_eq!(id_back, [3; 16]);
12951×
        assert_eq!(back.refs.len(), 2);
12961×
        assert_eq!(back.refs[0].attestation.as_ref().unwrap().attest, att);
12971×
        assert_eq!(back.refs[1].last_serial, None);
12981×
        assert_eq!(back.policy.as_ref().unwrap().policy, policy);
12991×
        assert_eq!(back.history, resp.history);
13001×
        assert_eq!(back.tags, vec![tag.clone()]);
1301
13021×
        let (h, w) = wire::decode_negotiate(&wire::encode_negotiate(
13031×
            &[ObjectRef([1; 32])],
13041×
            &[ObjectRef([2; 32])],
13051×
        ))
13061×
        .unwrap();
13071×
        assert_eq!((h, w), (vec![ObjectRef([1; 32])], vec![ObjectRef([2; 32])]));
13081×
        assert_eq!(
13091×
            wire::decode_hashes(&wire::encode_hashes(&[c.hash()])).unwrap(),
13101×
            vec![c.hash()]
1311
        );
13121×
        assert_eq!(
13131×
            wire::decode_fetch(&wire::encode_fetch(&[c.hash()], Some(4096))).unwrap(),
13141×
            (vec![c.hash()], Some(4096))
1315
        );
13161×
        assert_eq!(
13171×
            wire::decode_fetch(&wire::encode_fetch(&[], None)).unwrap(),
13181×
            (vec![], None)
1319
        );
13201×
        assert_eq!(
13211×
            wire::decode_shallow(&wire::encode_shallow(&[c.hash()], 3)).unwrap(),
13221×
            (vec![c.hash()], 3)
1323
        );
1324
        // An old-style fetch request without the filter field still decodes.
13251×
        assert_eq!(
13261×
            wire::decode_fetch(&wire::encode_hashes(&[c.hash()])).unwrap(),
13271×
            (vec![c.hash()], None)
1328
        );
1329
13301×
        let up = RefUpdate {
13311×
            name: "main".into(),
13321×
            old_state: ObjectRef([1; 32]),
13331×
            new_state: ObjectRef([2; 32]),
13341×
            force: true,
13351×
            attestation: Some(SignedAttest {
13361×
                attest: att,
13371×
                sigs: vec![sig],
13381×
            }),
13391×
        };
13401×
        let sp = SignedPolicy {
13411×
            policy,
13421×
            sigs: vec![psig],
13431×
        };
13441×
        let (b2, ups, pol, tags) = wire::decode_push(&wire::encode_push(
13451×
            &bundle,
13461×
            std::slice::from_ref(&up),
13471×
            Some(&sp),
13481×
            std::slice::from_ref(&tag),
13491×
        ))
13501×
        .unwrap();
13511×
        assert_eq!(pol, Some(sp));
13521×
        assert_eq!(tags, vec![tag]);
13531×
        assert_eq!(b2.changes.len(), 1);
13541×
        assert_eq!(ups[0].name, "main");
13551×
        assert!(ups[0].force);
13561×
        assert_eq!(ups[0].attestation, up.attestation);
13571×
        let res = vec![RefResult {
13581×
            name: "main".into(),
13591×
            ok: false,
13601×
            reason: "stale".into(),
13611×
        }];
13621×
        assert_eq!(
13631×
            wire::decode_push_response(&wire::encode_push_response(&res)).unwrap(),
1364
            res
1365
        );
13661×
        assert_eq!(
13671×
            wire::decode_capabilities(&wire::encode_capabilities([7; 16])).unwrap(),
1368
            [7; 16]
1369
        );
13701×
    }
1371
}