OXYGENHUB

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

810 of 972 lines run, 143 of 219 functions · run #313 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 {
1741011×
    pub fn open(path: &Path) -> Result<LocalRemote> {
175
        Ok(LocalRemote {
1761011×
            repo: Repo::open(path)?,
1771011×
            claimed_id: None,
178
        })
1791011×
    }
180
}
181
182
impl Remote for LocalRemote {
183794×
    fn repo_id(&self) -> Result<[u8; 16]> {
184794×
        self.repo.id()
185794×
    }
186793×
    fn refs(&self) -> Result<RefsResponse> {
187793×
        self.repo.remote_refs()
188793×
    }
189297×
    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>> {
190297×
        self.repo.negotiate(have, want)
191297×
    }
192300×
    fn fetch(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle> {
193300×
        self.repo.bundle_raw_for(changes, max_blob)
194300×
    }
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×
    }
204236×
    fn push(
205236×
        &self,
206236×
        bundle: Bundle,
207236×
        updates: Vec<RefUpdate>,
208236×
        policy: Option<SignedPolicy>,
209236×
        tags: Vec<TagRef>,
210236×
    ) -> Result<Vec<RefResult>> {
211236×
        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
            }
221225×
        }
222236×
        self.repo.receive_push(bundle, updates, policy, tags)
223236×
    }
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.
2671071×
pub fn open_store(
2681071×
    url: &str,
2691071×
    options: &RemoteOptions,
2701071×
) -> Option<crate::strata::Result<Box<dyn crate::strata::Store>>> {
2711071×
    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
        );
2761012×
    }
2771012×
    let openers = STORE_OPENERS.lock().unwrap_or_else(|p| p.into_inner());
2781012×
    openers.iter().find_map(|open| open(url, options))
2791071×
}
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.
299157×
pub fn tls_client_config(ca: Option<&Path>) -> Result<Arc<rustls::ClientConfig>> {
300157×
    let mut roots = rustls::RootCertStore::empty();
301157×
    roots.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
302157×
    if let Some(path) = ca {
30339×
        let pem = std::fs::read(path)
30439×
            .map_err(|e| RepoError::Msg(format!("reading CA file {}: {e}", path.display())))?;
30539×
        let certs = rustls_pemfile::certs(&mut pem.as_slice())
30639×
            .collect::<std::result::Result<Vec<_>, _>>()
30739×
            .map_err(|e| RepoError::Msg(format!("parsing CA file {}: {e}", path.display())))?;
30839×
        if certs.is_empty() {
3090×
            return Err(RepoError::Msg(format!(
3100×
                "CA file {} holds no certificates",
3110×
                path.display()
3120×
            )));
31339×
        }
31439×
        for c in certs {
31539×
            roots
31639×
                .add(c)
31739×
                .map_err(|e| RepoError::Msg(format!("CA file {}: {e}", path.display())))?;
318
        }
319118×
    }
320157×
    let cfg = rustls::ClientConfig::builder_with_provider(Arc::new(
321157×
        rustls::crypto::ring::default_provider(),
322
    ))
323157×
    .with_safe_default_protocol_versions()
324157×
    .map_err(|e| RepoError::Msg(e.to_string()))?
325157×
    .with_root_certificates(roots)
326157×
    .with_no_client_auth();
327157×
    Ok(Arc::new(cfg))
328157×
}
329
330
impl HttpRemote {
331153×
    pub fn new(base: &str, options: &RemoteOptions) -> Result<HttpRemote> {
332153×
        refuse_plain_token(base, options)?;
333153×
        let agent = ureq::AgentBuilder::new()
334153×
            .tls_config(tls_client_config(options.ca.as_deref())?)
335153×
            .build();
336153×
        Ok(HttpRemote {
337153×
            base: base.trim_end_matches('/').to_string(),
338153×
            token: Mutex::new(options.token.clone()),
339153×
            fallbacks: Mutex::new(options.fallback_tokens.iter().rev().cloned().collect()),
340153×
            repo_id: options.repo_id,
341153×
            agent,
342153×
        })
343153×
    }
344
345
    /// A request, tried again with the next fallback token while the
346
    /// hub refuses the one in use (oxygen#112).
347388×
    fn call(&self, post: bool, endpoint: &str, body: Option<&[u8]>) -> Result<Vec<u8>> {
348
        loop {
349389×
            let token = self.token.lock().unwrap_or_else(|p| p.into_inner()).clone();
350389×
            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
                }
366386×
                other => return other,
367
            }
368
        }
369388×
    }
370
371389×
    fn call_once(
372389×
        &self,
373389×
        post: bool,
374389×
        endpoint: &str,
375389×
        body: Option<&[u8]>,
376389×
        token: Option<&str>,
377389×
    ) -> Result<Vec<u8>> {
378389×
        let url = format!("{}/oxy/v1/{endpoint}", self.base);
379389×
        let mut req = if post {
380134×
            self.agent.post(&url)
381
        } else {
382255×
            self.agent.get(&url)
383
        };
384389×
        req = req.set("Accept", CONTENT_TYPE);
385389×
        if let Some(t) = token {
38622×
            req = req.set("Authorization", &format!("Bearer {t}"));
387367×
        }
388389×
        if let Some(id) = &self.repo_id {
389364×
            req = req.set(REPO_ID_HEADER, &hex::encode(id));
390364×
        }
391389×
        let resp = match body {
392134×
            Some(b) => req.set("Content-Type", CONTENT_TYPE).send_bytes(b),
393255×
            None => req.call(),
394
        };
39511×
        match resp {
396378×
            Ok(r) => {
397378×
                let mut out = Vec::new();
398378×
                r.into_reader()
399378×
                    .read_to_end(&mut out)
400378×
                    .map_err(|e| RepoError::Msg(format!("{url}: {e}")))?;
401378×
                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
        }
411389×
    }
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`]).
425153×
fn refuse_plain_token(url: &str, options: &RemoteOptions) -> Result<()> {
426153×
    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×
        )));
430153×
    }
431153×
    Ok(())
432153×
}
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 {
536132×
    fn repo_id(&self) -> Result<[u8; 16]> {
537132×
        wire::decode_capabilities(&self.call(false, "capabilities", None)?)
538132×
    }
539123×
    fn refs(&self) -> Result<RefsResponse> {
540123×
        Ok(wire::decode_refs(&self.call(false, "refs", None)?)?.1)
541123×
    }
54249×
    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>> {
54349×
        wire::decode_hashes(&self.call(
544
            true,
54549×
            "negotiate",
54649×
            Some(&wire::encode_negotiate(have, want)),
5470×
        )?)
54849×
    }
54949×
    fn fetch(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle> {
55049×
        wire::decode_bundle(&self.call(
551
            true,
55249×
            "fetch",
55349×
            Some(&wire::encode_fetch(changes, max_blob)),
5540×
        )?)
55549×
    }
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×
    }
57234×
    fn push(
57334×
        &self,
57434×
        bundle: Bundle,
57534×
        updates: Vec<RefUpdate>,
57634×
        policy: Option<SignedPolicy>,
57734×
        tags: Vec<TagRef>,
57834×
    ) -> Result<Vec<RefResult>> {
57934×
        wire::decode_push_response(&self.call(
580
            true,
58134×
            "push",
58234×
            Some(&wire::encode_push(
58334×
                &bundle,
58434×
                &updates,
58534×
                policy.as_ref(),
58634×
                &tags,
58734×
            )),
5884×
        )?)
58934×
    }
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).
595248×
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.
598248×
    let (prefix, path) = match (url.strip_prefix("file://"), url.strip_prefix("dir://")) {
5991×
        (Some(p), _) => ("file://", p),
60011×
        (_, Some(p)) => ("dir://", p),
601236×
        _ if url.contains("://") => return url.to_string(),
602213×
        _ => ("", url),
603
    };
604225×
    match std::path::absolute(path) {
605225×
        Ok(abs) => format!("{prefix}{}", abs.display()),
6060×
        Err(_) => url.to_string(),
607
    }
608248×
}
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).
6121212×
pub fn open_remote(url: &str, options: &RemoteOptions) -> Result<Box<dyn Remote>> {
6131212×
    if url.starts_with("http://") || url.starts_with("https://") {
614153×
        return Ok(Box::new(HttpRemote::new(url, options)?));
6151059×
    }
6161059×
    if let Some(store) = open_store(url, options) {
61748×
        return crate::repo::strata::open_store_remote(store?, options);
6181011×
    }
6191011×
    let path = url.strip_prefix("file://").unwrap_or(url);
6201011×
    if path.contains("://") {
6210×
        return Err(RepoError::Msg(format!("unsupported remote URL {url:?}")));
6221011×
    }
6231011×
    let mut local = LocalRemote::open(Path::new(path))?;
6241011×
    local.claimed_id = options.repo_id;
6251011×
    Ok(Box::new(local))
6261212×
}
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×
    }
637147×
    fn sigs_value(sigs: &[Signature]) -> Value {
638147×
        Value::Array(sigs.iter().map(|s| s.to_value()).collect())
639147×
    }
640142×
    fn sigs_from(v: &Value) -> Result<Vec<Signature>> {
641142×
        Ok(o::array(v, "signatures")?
642142×
            .iter()
643142×
            .map(Signature::from_value)
644142×
            .collect::<std::result::Result<Vec<_>, _>>()?)
645142×
    }
646366×
    fn hashes_value(hs: &[ObjectRef]) -> Value {
647535×
        Value::Array(hs.iter().map(|h| Value::Bytes(h.0.to_vec())).collect())
648366×
    }
649425×
    fn hashes_from(v: &Value) -> Result<Vec<ObjectRef>> {
650425×
        Ok(o::array(v, "hashes")?
651425×
            .iter()
652595×
            .map(|h| o::oref(h, "hash"))
653425×
            .collect::<std::result::Result<Vec<_>, _>>()?)
654425×
    }
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×
    }
668197×
    fn opt_attest_value(sa: &Option<SignedAttest>) -> Value {
669197×
        sa.as_ref().map(attest_value).unwrap_or(Value::Null)
670197×
    }
671252×
    fn opt_attest_from(v: &Value) -> Result<Option<SignedAttest>> {
672252×
        if v.is_null() {
673225×
            Ok(None)
674
        } else {
67527×
            attest_from(v).map(Some)
676
        }
677252×
    }
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×
    }
693186×
    fn opt_policy_value(sp: Option<&SignedPolicy>) -> Value {
694186×
        sp.map(policy_value).unwrap_or(Value::Null)
695186×
    }
696225×
    fn opt_policy_from(v: &Value) -> Result<Option<SignedPolicy>> {
697225×
        if v.is_null() {
698222×
            Ok(None)
699
        } else {
7003×
            policy_from(v).map(Some)
701
        }
702225×
    }
703931×
    fn map<'a>(v: &'a Value, what: &str) -> Result<&'a o::Map> {
704931×
        Ok(o::map(v, what)?)
705931×
    }
7062603×
    fn field<'a>(m: &'a o::Map, k: u64, what: &str) -> Result<&'a Value> {
7072603×
        Ok(o::field(m, k, what)?)
7082603×
    }
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.
713120×
    pub fn bundle_value(b: &Bundle) -> Value {
714120×
        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.
717120×
        for (bytes, s) in &b.raw_changes {
7180×
            changes.push(Value::Array(vec![
7190×
                Value::Bytes(bytes.clone()),
7200×
                sigs_value(s),
7210×
            ]));
7220×
        }
723120×
        let mut texts: Vec<(ObjectType, Vec<u8>)> = Vec::new();
724120×
        let mut listed: Option<std::collections::HashSet<ObjectRef>> = None;
725120×
        for (c, s) in &b.changes {
726113×
            let (bytes, stored) = c.encode_parts();
727113×
            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
            }
744113×
            changes.push(Value::Array(vec![Value::Bytes(bytes), sigs_value(s)]));
745
        }
746120×
        let mut v = Value::map([
747120×
            (0, Value::Array(changes)),
748120×
            (
749120×
                1,
750120×
                Value::Array(b.attests.iter().map(attest_value).collect()),
751120×
            ),
752
            (
753
                2,
754
                Value::Array(
755120×
                    b.objects
756120×
                        .iter()
757120×
                        .chain(&texts)
758120×
                        .map(|(t, body)| {
75988×
                            Value::Array(vec![
76088×
                                Value::Text(t.name().into()),
76188×
                                Value::Bytes(body.clone()),
76288×
                            ])
76388×
                        })
764120×
                        .collect(),
765
                ),
766
            ),
767120×
            (
768120×
                3,
769120×
                Value::Array(b.policies.iter().map(policy_value).collect()),
770120×
            ),
771
        ]);
772
        // Only when there are any; older peers ignore the field.
773120×
        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×
            }
785119×
        }
786120×
        v
787120×
    }
788
789110×
    pub fn bundle_from_value(v: &Value) -> Result<Bundle> {
790110×
        let m = map(v, "bundle")?;
791110×
        let mut b = Bundle::default();
792110×
        for ob in o::array(field(m, 2, "objects")?, "objects")? {
79377×
            let pair = o::array(ob, "object")?;
79477×
            if pair.len() != 2 {
7950×
                return Err(schema("object: expected [type, body]"));
79677×
            }
79777×
            let ty = ObjectType::from_name(o::text(&pair[0], "object type")?)
79877×
                .ok_or_else(|| schema("unknown object type"))?;
79977×
            b.objects
80077×
                .push((ty, o::bytes(&pair[1], "object body")?.to_vec()));
801
        }
802
        // Stored inserts are read from the bundle's own objects.
803110×
        let mut by_ref: Option<std::collections::HashMap<ObjectRef, usize>> = None;
804110×
        for c in o::array(field(m, 0, "changes")?, "changes")? {
80596×
            let pair = o::array(c, "change entry")?;
80696×
            if pair.len() != 2 {
8070×
                return Err(schema("change entry: expected [change, sigs]"));
80896×
            }
80996×
            let encoded = o::bytes(&pair[0], "change")?;
81096×
            b.hashes
81196×
                .push(crate::hash::hash_object(ObjectType::Change, encoded));
81296×
            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×
            })?;
83696×
            b.changes.push((change, sigs_from(&pair[1])?));
837
        }
838110×
        for a in o::array(field(m, 1, "attests")?, "attests")? {
83911×
            b.attests.push(attest_from(a)?);
840
        }
841110×
        if let Some(ps) = m.get(&3) {
842110×
            for p in o::array(ps, "policies")? {
8433×
                b.policies.push(policy_from(p)?);
844
            }
8450×
        }
846110×
        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
            }
853109×
        }
854110×
        Ok(b)
855110×
    }
856
85785×
    pub fn encode_bundle(b: &Bundle) -> Vec<u8> {
85885×
        cbor::encode(&bundle_value(b))
85985×
    }
86079×
    pub fn decode_bundle(bytes: &[u8]) -> Result<Bundle> {
86179×
        bundle_from_value(&cbor::decode(bytes).map_err(ObjectError::from)?)
86279×
    }
863
864127×
    pub fn encode_capabilities(id: [u8; 16]) -> Vec<u8> {
865127×
        cbor::encode(&Value::map([
866127×
            (0, Value::Array(vec![Value::Uint(PROTOCOL_VERSION)])),
867127×
            (1, Value::Bytes(id.to_vec())),
868127×
            (2, Value::Array(vec![Value::Text("seal".into())])),
869127×
        ]))
870127×
    }
871127×
    pub fn decode_capabilities(bytes: &[u8]) -> Result<[u8; 16]> {
872127×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
873127×
        let m = map(&v, "capabilities")?;
874127×
        let versions = o::array(field(m, 0, "versions")?, "versions")?;
875127×
        if !versions
876127×
            .iter()
877127×
            .any(|x| x.as_uint() == Some(PROTOCOL_VERSION))
878
        {
8790×
            return Err(RepoError::Msg(
8800×
                "remote does not speak protocol version 1".into(),
8810×
            ));
882127×
        }
883127×
        o::bytes(field(m, 1, "repo id")?, "repo id")?
884127×
            .try_into()
885127×
            .map_err(|_| schema("repo id: expected 16 bytes"))
886127×
    }
887
888186×
    fn tags_value(tags: &[TagRef]) -> Value {
889
        Value::Array(
890186×
            tags.iter()
891186×
                .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×
                })
899186×
                .collect(),
900
        )
901186×
    }
902225×
    fn tags_from(v: &Value) -> Result<Vec<TagRef>> {
903225×
        let mut out = Vec::new();
904225×
        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
        }
913225×
        Ok(out)
914225×
    }
915
916151×
    pub fn encode_refs(id: [u8; 16], resp: &RefsResponse) -> Vec<u8> {
917151×
        cbor::encode(&Value::map([
918
            (
919
                0,
920
                Value::Array(
921151×
                    resp.refs
922151×
                        .iter()
923155×
                        .map(|r| {
924155×
                            let mut v = Value::map([
925155×
                                (0, Value::Text(r.name.clone())),
926155×
                                (1, Value::Bytes(r.state.0.to_vec())),
927155×
                                (2, hashes_value(&r.heads)),
928155×
                                (3, opt_attest_value(&r.attestation)),
929155×
                                (4, r.last_serial.map(Value::Uint).unwrap_or(Value::Null)),
930155×
                            ]);
931
                            // Only when there is one; older peers ignore it.
932155×
                            if r.accepted.is_some() {
9330×
                                if let Value::Map(m) = &mut v {
9340×
                                    m.insert(5, opt_attest_value(&r.accepted));
9350×
                                }
936155×
                            }
937155×
                            v
938155×
                        })
939151×
                        .collect(),
940
                ),
941
            ),
942151×
            (1, Value::Bytes(id.to_vec())),
943151×
            (2, opt_policy_value(resp.policy.as_ref())),
944151×
            (3, tags_value(&resp.tags)),
945151×
            (
946151×
                4,
947151×
                Value::Array(resp.history.iter().map(policy_value).collect()),
948151×
            ),
949151×
            (
950151×
                5,
951151×
                resp.default_bookmark
952151×
                    .clone()
953151×
                    .map(Value::Text)
954151×
                    .unwrap_or(Value::Null),
955151×
            ),
956
        ]))
957151×
    }
958194×
    pub fn decode_refs(bytes: &[u8]) -> Result<([u8; 16], RefsResponse)> {
959194×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
960194×
        let m = map(&v, "refs")?;
961194×
        let id: [u8; 16] = o::bytes(field(m, 1, "repo id")?, "repo id")?
962194×
            .try_into()
963194×
            .map_err(|_| schema("repo id: expected 16 bytes"))?;
964194×
        let mut refs = Vec::new();
965198×
        for r in o::array(field(m, 0, "refs")?, "refs")? {
966198×
            let rm = map(r, "ref")?;
967198×
            let ls = field(rm, 4, "last serial")?;
968198×
            refs.push(RemoteRef {
969198×
                name: o::text(field(rm, 0, "name")?, "name")?.to_owned(),
970198×
                state: o::oref(field(rm, 1, "state")?, "state")?,
971198×
                heads: hashes_from(field(rm, 2, "heads")?)?,
972198×
                attestation: opt_attest_from(field(rm, 3, "attestation")?)?,
973198×
                last_serial: if ls.is_null() {
974182×
                    None
975
                } else {
97616×
                    Some(o::uint(ls, "last serial")?)
977
                },
978198×
                accepted: match rm.get(&5) {
9790×
                    Some(a) => opt_attest_from(a)?,
980198×
                    None => None,
981
                },
982
            });
983
        }
984194×
        let policy = match m.get(&2) {
985194×
            Some(p) => opt_policy_from(p)?,
9860×
            None => None,
987
        };
988194×
        let tags = match m.get(&3) {
989194×
            Some(t) => tags_from(t)?,
9900×
            None => Vec::new(),
991
        };
992194×
        let mut history = Vec::new();
993194×
        if let Some(h) = m.get(&4) {
994194×
            for p in o::array(h, "history")? {
9951×
                history.push(policy_from(p)?);
996
            }
9970×
        }
998194×
        let default_bookmark = match m.get(&5) {
999125×
            Some(Value::Text(t)) => Some(t.clone()),
100069×
            _ => None,
1001
        };
1002194×
        Ok((
1003194×
            id,
1004194×
            RefsResponse {
1005194×
                refs,
1006194×
                policy,
1007194×
                tags,
1008194×
                history,
1009194×
                default_bookmark,
1010194×
            },
1011194×
        ))
1012194×
    }
1013
101450×
    pub fn encode_negotiate(have: &[ChangeHash], want: &[ChangeHash]) -> Vec<u8> {
101550×
        cbor::encode(&Value::map([
101650×
            (0, hashes_value(have)),
101750×
            (1, hashes_value(want)),
101850×
        ]))
101950×
    }
102050×
    pub fn decode_negotiate(bytes: &[u8]) -> Result<(Vec<ChangeHash>, Vec<ChangeHash>)> {
102150×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
102250×
        let m = map(&v, "negotiate")?;
1023
        Ok((
102450×
            hashes_from(field(m, 0, "have")?)?,
102550×
            hashes_from(field(m, 1, "want")?)?,
1026
        ))
102750×
    }
1028
1029
    /// `fetch` request: `{0: [changes], 1: max blob size or null}`.
103051×
    pub fn encode_fetch(hs: &[ChangeHash], max_blob: Option<u64>) -> Vec<u8> {
103151×
        cbor::encode(&Value::map([
103251×
            (0, hashes_value(hs)),
103351×
            (1, max_blob.map(Value::Uint).unwrap_or(Value::Null)),
103451×
        ]))
103551×
    }
103652×
    pub fn decode_fetch(bytes: &[u8]) -> Result<(Vec<ChangeHash>, Option<u64>)> {
103752×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
103852×
        let m = map(&v, "fetch")?;
103952×
        let max = match m.get(&1) {
104051×
            Some(x) if !x.is_null() => Some(o::uint(x, "max blob")?),
104150×
            _ => None,
1042
        };
104352×
        Ok((hashes_from(field(m, 0, "changes")?)?, max))
104452×
    }
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
106252×
    pub fn encode_hashes(hs: &[ChangeHash]) -> Vec<u8> {
106352×
        cbor::encode(&Value::map([(0, hashes_value(hs))]))
106452×
    }
106551×
    pub fn decode_hashes(bytes: &[u8]) -> Result<Vec<ChangeHash>> {
106651×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
106751×
        hashes_from(field(map(&v, "hashes")?, 0, "hashes")?)
106851×
    }
1069
107035×
    pub fn encode_push(
107135×
        bundle: &Bundle,
107235×
        updates: &[RefUpdate],
107335×
        policy: Option<&SignedPolicy>,
107435×
        tags: &[TagRef],
107535×
    ) -> Vec<u8> {
107635×
        cbor::encode(&Value::map([
107735×
            (0, bundle_value(bundle)),
107835×
            (2, opt_policy_value(policy)),
107935×
            (3, tags_value(tags)),
1080
            (
1081
                1,
1082
                Value::Array(
108335×
                    updates
108435×
                        .iter()
108535×
                        .map(|u| {
108635×
                            Value::map([
108735×
                                (0, Value::Text(u.name.clone())),
108835×
                                (1, Value::Bytes(u.old_state.0.to_vec())),
108935×
                                (2, Value::Bytes(u.new_state.0.to_vec())),
109035×
                                (3, Value::Bool(u.force)),
109135×
                                (4, opt_attest_value(&u.attestation)),
109235×
                            ])
109335×
                        })
109435×
                        .collect(),
1095
                ),
1096
            ),
1097
        ]))
109835×
    }
1099
    /// A decoded push: bundle, bookmark updates, policy, tags.
1100
    pub type PushParts = (Bundle, Vec<RefUpdate>, Option<SignedPolicy>, Vec<TagRef>);
1101
110231×
    pub fn decode_push(bytes: &[u8]) -> Result<PushParts> {
110331×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
110431×
        let m = map(&v, "push")?;
110531×
        let bundle = bundle_from_value(field(m, 0, "bundle")?)?;
110631×
        let mut updates = Vec::new();
110731×
        for u in o::array(field(m, 1, "updates")?, "updates")? {
110831×
            let um = map(u, "update")?;
110931×
            updates.push(RefUpdate {
111031×
                name: o::text(field(um, 0, "name")?, "name")?.to_owned(),
111131×
                old_state: o::oref(field(um, 1, "old_state")?, "old_state")?,
111231×
                new_state: o::oref(field(um, 2, "new_state")?, "new_state")?,
111331×
                force: field(um, 3, "force")?
111431×
                    .as_bool()
111531×
                    .ok_or_else(|| schema("force: expected bool"))?,
111631×
                attestation: opt_attest_from(field(um, 4, "attestation")?)?,
1117
            });
1118
        }
111931×
        let policy = match m.get(&2) {
112031×
            Some(p) => opt_policy_from(p)?,
11210×
            None => None,
1122
        };
112331×
        let tags = match m.get(&3) {
112431×
            Some(t) => tags_from(t)?,
11250×
            None => Vec::new(),
1126
        };
112731×
        Ok((bundle, updates, policy, tags))
112831×
    }
1129
113031×
    pub fn encode_push_response(results: &[RefResult]) -> Vec<u8> {
113131×
        cbor::encode(&Value::map([(
1132
            0,
1133
            Value::Array(
113431×
                results
113531×
                    .iter()
113632×
                    .map(|r| {
113732×
                        Value::map([
113832×
                            (0, Value::Text(r.name.clone())),
113932×
                            (1, Value::Bool(r.ok)),
114032×
                            (2, Value::Text(r.reason.clone())),
114132×
                        ])
114232×
                    })
114331×
                    .collect(),
1144
            ),
1145
        )]))
114631×
    }
114731×
    pub fn decode_push_response(bytes: &[u8]) -> Result<Vec<RefResult>> {
114831×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
114931×
        let m = map(&v, "push response")?;
115031×
        let mut out = Vec::new();
115132×
        for r in o::array(field(m, 0, "results")?, "results")? {
115232×
            let rm = map(r, "result")?;
115332×
            out.push(RefResult {
115432×
                name: o::text(field(rm, 0, "name")?, "name")?.to_owned(),
115532×
                ok: field(rm, 1, "ok")?
115632×
                    .as_bool()
115732×
                    .ok_or_else(|| schema("ok: expected bool"))?,
115832×
                reason: o::text(field(rm, 2, "reason")?, "reason")?.to_owned(),
1159
            });
1160
        }
116131×
        Ok(out)
116231×
    }
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×
            },
12301×
            &issuer,
12311×
            &issuer_sk,
1232
        ));
12331×
        let encoded = wire::encode_bundle(&with);
12341×
        assert_ne!(encoded, plain);
12351×
        assert_eq!(
12361×
            wire::decode_bundle(&encoded).unwrap().certificates,
1237
            with.certificates
1238
        );
12391×
        with.certificates.clear();
12401×
        assert_eq!(wire::encode_bundle(&with), plain);
1241
12421×
        let att = Attest {
12431×
            repo: [3; 16],
12441×
            hlc: Hlc::default(),
12451×
            body: AttestBody::Ref {
12461×
                name: "main".into(),
12471×
                old_state: ObjectRef([1; 32]),
12481×
                new_state: ObjectRef([2; 32]),
12491×
                serial: 4,
12501×
            },
12511×
        };
12521×
        let refs = vec![
12531×
            RemoteRef {
12541×
                name: "main".into(),
12551×
                state: ObjectRef([2; 32]),
12561×
                heads: vec![c.hash()],
12571×
                attestation: Some(SignedAttest {
12581×
                    attest: att.clone(),
12591×
                    sigs: vec![sig.clone()],
12601×
                }),
12611×
                last_serial: Some(4),
12621×
                accepted: None,
12631×
            },
12641×
            RemoteRef {
12651×
                name: "dev".into(),
12661×
                state: ObjectRef([5; 32]),
12671×
                heads: vec![],
12681×
                attestation: None,
12691×
                last_serial: None,
12701×
                accepted: None,
12711×
            },
1272
        ];
12731×
        let tag = TagRef {
12741×
            name: "v1".into(),
12751×
            state: ObjectRef([2; 32]),
12761×
            heads: vec![c.hash()],
12771×
            attestation: None,
12781×
        };
12791×
        let resp = RefsResponse {
12801×
            refs,
12811×
            policy: Some(SignedPolicy {
12821×
                policy: policy.clone(),
12831×
                sigs: vec![psig.clone()],
12841×
            }),
12851×
            tags: vec![tag.clone()],
12861×
            history: vec![SignedPolicy {
12871×
                policy: policy.clone(),
12881×
                sigs: vec![],
12891×
            }],
12901×
            default_bookmark: Some("develop".into()),
12911×
        };
12921×
        let (id_back, back) = wire::decode_refs(&wire::encode_refs([3; 16], &resp)).unwrap();
12931×
        assert_eq!(id_back, [3; 16]);
12941×
        assert_eq!(back.refs.len(), 2);
12951×
        assert_eq!(back.refs[0].attestation.as_ref().unwrap().attest, att);
12961×
        assert_eq!(back.refs[1].last_serial, None);
12971×
        assert_eq!(back.policy.as_ref().unwrap().policy, policy);
12981×
        assert_eq!(back.history, resp.history);
12991×
        assert_eq!(back.tags, vec![tag.clone()]);
1300
13011×
        let (h, w) = wire::decode_negotiate(&wire::encode_negotiate(
13021×
            &[ObjectRef([1; 32])],
13031×
            &[ObjectRef([2; 32])],
13041×
        ))
13051×
        .unwrap();
13061×
        assert_eq!((h, w), (vec![ObjectRef([1; 32])], vec![ObjectRef([2; 32])]));
13071×
        assert_eq!(
13081×
            wire::decode_hashes(&wire::encode_hashes(&[c.hash()])).unwrap(),
13091×
            vec![c.hash()]
1310
        );
13111×
        assert_eq!(
13121×
            wire::decode_fetch(&wire::encode_fetch(&[c.hash()], Some(4096))).unwrap(),
13131×
            (vec![c.hash()], Some(4096))
1314
        );
13151×
        assert_eq!(
13161×
            wire::decode_fetch(&wire::encode_fetch(&[], None)).unwrap(),
13171×
            (vec![], None)
1318
        );
13191×
        assert_eq!(
13201×
            wire::decode_shallow(&wire::encode_shallow(&[c.hash()], 3)).unwrap(),
13211×
            (vec![c.hash()], 3)
1322
        );
1323
        // An old-style fetch request without the filter field still decodes.
13241×
        assert_eq!(
13251×
            wire::decode_fetch(&wire::encode_hashes(&[c.hash()])).unwrap(),
13261×
            (vec![c.hash()], None)
1327
        );
1328
13291×
        let up = RefUpdate {
13301×
            name: "main".into(),
13311×
            old_state: ObjectRef([1; 32]),
13321×
            new_state: ObjectRef([2; 32]),
13331×
            force: true,
13341×
            attestation: Some(SignedAttest {
13351×
                attest: att,
13361×
                sigs: vec![sig],
13371×
            }),
13381×
        };
13391×
        let sp = SignedPolicy {
13401×
            policy,
13411×
            sigs: vec![psig],
13421×
        };
13431×
        let (b2, ups, pol, tags) = wire::decode_push(&wire::encode_push(
13441×
            &bundle,
13451×
            std::slice::from_ref(&up),
13461×
            Some(&sp),
13471×
            std::slice::from_ref(&tag),
13481×
        ))
13491×
        .unwrap();
13501×
        assert_eq!(pol, Some(sp));
13511×
        assert_eq!(tags, vec![tag]);
13521×
        assert_eq!(b2.changes.len(), 1);
13531×
        assert_eq!(ups[0].name, "main");
13541×
        assert!(ups[0].force);
13551×
        assert_eq!(ups[0].attestation, up.attestation);
13561×
        let res = vec![RefResult {
13571×
            name: "main".into(),
13581×
            ok: false,
13591×
            reason: "stale".into(),
13601×
        }];
13611×
        assert_eq!(
13621×
            wire::decode_push_response(&wire::encode_push_response(&res)).unwrap(),
1363
            res
1364
        );
13651×
        assert_eq!(
13661×
            wire::decode_capabilities(&wire::encode_capabilities([7; 16])).unwrap(),
1367
            [7; 16]
1368
        );
13691×
    }
1370
}