OXYGENHUB

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

811 of 973 lines run, 143 of 219 functions · run #319 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×
    }
204246×
    fn push(
205246×
        &self,
206246×
        bundle: Bundle,
207246×
        updates: Vec<RefUpdate>,
208246×
        policy: Option<SignedPolicy>,
209246×
        tags: Vec<TagRef>,
210246×
    ) -> Result<Vec<RefResult>> {
211246×
        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
            }
221235×
        }
222246×
        self.repo.receive_push(bundle, updates, policy, tags)
223246×
    }
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.
299153×
pub fn tls_client_config(ca: Option<&Path>) -> Result<Arc<rustls::ClientConfig>> {
300153×
    let mut roots = rustls::RootCertStore::empty();
301153×
    roots.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
302153×
    if let Some(path) = ca {
30365×
        let pem = std::fs::read(path)
30465×
            .map_err(|e| RepoError::Msg(format!("reading CA file {}: {e}", path.display())))?;
30565×
        let certs = rustls_pemfile::certs(&mut pem.as_slice())
30665×
            .collect::<std::result::Result<Vec<_>, _>>()
30765×
            .map_err(|e| RepoError::Msg(format!("parsing CA file {}: {e}", path.display())))?;
30865×
        if certs.is_empty() {
3090×
            return Err(RepoError::Msg(format!(
3100×
                "CA file {} holds no certificates",
3110×
                path.display()
3120×
            )));
31365×
        }
31465×
        for c in certs {
31565×
            roots
31665×
                .add(c)
31765×
                .map_err(|e| RepoError::Msg(format!("CA file {}: {e}", path.display())))?;
318
        }
31988×
    }
320153×
    let cfg = rustls::ClientConfig::builder_with_provider(Arc::new(
321153×
        rustls::crypto::ring::default_provider(),
322
    ))
323153×
    .with_safe_default_protocol_versions()
324153×
    .map_err(|e| RepoError::Msg(e.to_string()))?
325153×
    .with_root_certificates(roots)
326153×
    .with_no_client_auth();
327153×
    Ok(Arc::new(cfg))
328153×
}
329
330
impl HttpRemote {
331149×
    pub fn new(base: &str, options: &RemoteOptions) -> Result<HttpRemote> {
332149×
        refuse_plain_token(base, options)?;
333149×
        let agent = ureq::AgentBuilder::new()
334149×
            .tls_config(tls_client_config(options.ca.as_deref())?)
335149×
            .build();
336149×
        Ok(HttpRemote {
337149×
            base: base.trim_end_matches('/').to_string(),
338149×
            token: Mutex::new(options.token.clone()),
339149×
            fallbacks: Mutex::new(options.fallback_tokens.iter().rev().cloned().collect()),
340149×
            repo_id: options.repo_id,
341149×
            agent,
342149×
        })
343149×
    }
344
345
    /// A request, tried again with the next fallback token while the
346
    /// hub refuses the one in use (oxygen#112).
347380×
    fn call(&self, post: bool, endpoint: &str, body: Option<&[u8]>) -> Result<Vec<u8>> {
348
        loop {
349381×
            let token = self.token.lock().unwrap_or_else(|p| p.into_inner()).clone();
350381×
            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
                }
366378×
                other => return other,
367
            }
368
        }
369380×
    }
370
371381×
    fn call_once(
372381×
        &self,
373381×
        post: bool,
374381×
        endpoint: &str,
375381×
        body: Option<&[u8]>,
376381×
        token: Option<&str>,
377381×
    ) -> Result<Vec<u8>> {
378381×
        let url = format!("{}/oxy/v1/{endpoint}", self.base);
379381×
        let mut req = if post {
380134×
            self.agent.post(&url)
381
        } else {
382247×
            self.agent.get(&url)
383
        };
384381×
        req = req.set("Accept", CONTENT_TYPE);
385381×
        if let Some(t) = token {
38622×
            req = req.set("Authorization", &format!("Bearer {t}"));
387359×
        }
388381×
        if let Some(id) = &self.repo_id {
389356×
            req = req.set(REPO_ID_HEADER, &hex::encode(id));
390356×
        }
391381×
        let resp = match body {
392134×
            Some(b) => req.set("Content-Type", CONTENT_TYPE).send_bytes(b),
393247×
            None => req.call(),
394
        };
39511×
        match resp {
396370×
            Ok(r) => {
397370×
                let mut out = Vec::new();
398370×
                r.into_reader()
399370×
                    .read_to_end(&mut out)
400370×
                    .map_err(|e| RepoError::Msg(format!("{url}: {e}")))?;
401370×
                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
        }
411381×
    }
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`]).
425149×
fn refuse_plain_token(url: &str, options: &RemoteOptions) -> Result<()> {
426149×
    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×
        )));
430149×
    }
431149×
    Ok(())
432149×
}
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 {
536128×
    fn repo_id(&self) -> Result<[u8; 16]> {
537128×
        wire::decode_capabilities(&self.call(false, "capabilities", None)?)
538128×
    }
539119×
    fn refs(&self) -> Result<RefsResponse> {
540119×
        Ok(wire::decode_refs(&self.call(false, "refs", None)?)?.1)
541119×
    }
54247×
    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>> {
54347×
        wire::decode_hashes(&self.call(
544
            true,
54547×
            "negotiate",
54647×
            Some(&wire::encode_negotiate(have, want)),
5470×
        )?)
54847×
    }
54947×
    fn fetch(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle> {
55047×
        wire::decode_bundle(&self.call(
551
            true,
55247×
            "fetch",
55347×
            Some(&wire::encode_fetch(changes, max_blob)),
5540×
        )?)
55547×
    }
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×
    }
57238×
    fn push(
57338×
        &self,
57438×
        bundle: Bundle,
57538×
        updates: Vec<RefUpdate>,
57638×
        policy: Option<SignedPolicy>,
57738×
        tags: Vec<TagRef>,
57838×
    ) -> Result<Vec<RefResult>> {
57938×
        wire::decode_push_response(&self.call(
580
            true,
58138×
            "push",
58238×
            Some(&wire::encode_push(
58338×
                &bundle,
58438×
                &updates,
58538×
                policy.as_ref(),
58638×
                &tags,
58738×
            )),
5884×
        )?)
58938×
    }
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).
6121260×
pub fn open_remote(url: &str, options: &RemoteOptions) -> Result<Box<dyn Remote>> {
6131260×
    if url.starts_with("http://") || url.starts_with("https://") {
614149×
        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))
6261260×
}
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×
    }
637151×
    fn sigs_value(sigs: &[Signature]) -> Value {
638151×
        Value::Array(sigs.iter().map(|s| s.to_value()).collect())
639151×
    }
640146×
    fn sigs_from(v: &Value) -> Result<Vec<Signature>> {
641146×
        Ok(o::array(v, "signatures")?
642146×
            .iter()
643146×
            .map(Signature::from_value)
644146×
            .collect::<std::result::Result<Vec<_>, _>>()?)
645146×
    }
646354×
    fn hashes_value(hs: &[ObjectRef]) -> Value {
647497×
        Value::Array(hs.iter().map(|h| Value::Bytes(h.0.to_vec())).collect())
648354×
    }
649413×
    fn hashes_from(v: &Value) -> Result<Vec<ObjectRef>> {
650413×
        Ok(o::array(v, "hashes")?
651413×
            .iter()
652557×
            .map(|h| o::oref(h, "hash"))
653413×
            .collect::<std::result::Result<Vec<_>, _>>()?)
654413×
    }
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×
    }
7062609×
    fn field<'a>(m: &'a o::Map, k: u64, what: &str) -> Result<&'a Value> {
7072609×
        Ok(o::field(m, k, what)?)
7082609×
    }
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.
713122×
    pub fn bundle_value(b: &Bundle) -> Value {
714122×
        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.
717122×
        for (bytes, s) in &b.raw_changes {
7180×
            changes.push(Value::Array(vec![
7190×
                Value::Bytes(bytes.clone()),
7200×
                sigs_value(s),
7210×
            ]));
7220×
        }
723122×
        let mut texts: Vec<(ObjectType, Vec<u8>)> = Vec::new();
724122×
        let mut listed: Option<std::collections::HashSet<ObjectRef>> = None;
725122×
        for (c, s) in &b.changes {
726117×
            let (bytes, stored) = c.encode_parts();
727117×
            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
            }
744117×
            changes.push(Value::Array(vec![Value::Bytes(bytes), sigs_value(s)]));
745
        }
746122×
        let mut v = Value::map([
747122×
            (0, Value::Array(changes)),
748122×
            (
749122×
                1,
750122×
                Value::Array(b.attests.iter().map(attest_value).collect()),
751122×
            ),
752
            (
753
                2,
754
                Value::Array(
755122×
                    b.objects
756122×
                        .iter()
757122×
                        .chain(&texts)
758122×
                        .map(|(t, body)| {
75992×
                            Value::Array(vec![
76092×
                                Value::Text(t.name().into()),
76192×
                                Value::Bytes(body.clone()),
76292×
                            ])
76392×
                        })
764122×
                        .collect(),
765
                ),
766
            ),
767122×
            (
768122×
                3,
769122×
                Value::Array(b.policies.iter().map(policy_value).collect()),
770122×
            ),
771
        ]);
772
        // Only when there are any; older peers ignore the field.
773122×
        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×
            }
785121×
        }
786122×
        v
787122×
    }
788
789112×
    pub fn bundle_from_value(v: &Value) -> Result<Bundle> {
790112×
        let m = map(v, "bundle")?;
791112×
        let mut b = Bundle::default();
792112×
        for ob in o::array(field(m, 2, "objects")?, "objects")? {
79381×
            let pair = o::array(ob, "object")?;
79481×
            if pair.len() != 2 {
7950×
                return Err(schema("object: expected [type, body]"));
79681×
            }
79781×
            let ty = ObjectType::from_name(o::text(&pair[0], "object type")?)
79881×
                .ok_or_else(|| schema("unknown object type"))?;
79981×
            b.objects
80081×
                .push((ty, o::bytes(&pair[1], "object body")?.to_vec()));
801
        }
802
        // Stored inserts are read from the bundle's own objects.
803112×
        let mut by_ref: Option<std::collections::HashMap<ObjectRef, usize>> = None;
804112×
        for c in o::array(field(m, 0, "changes")?, "changes")? {
805100×
            let pair = o::array(c, "change entry")?;
806100×
            if pair.len() != 2 {
8070×
                return Err(schema("change entry: expected [change, sigs]"));
808100×
            }
809100×
            let encoded = o::bytes(&pair[0], "change")?;
810100×
            b.hashes
811100×
                .push(crate::hash::hash_object(ObjectType::Change, encoded));
812100×
            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×
            })?;
836100×
            b.changes.push((change, sigs_from(&pair[1])?));
837
        }
838112×
        for a in o::array(field(m, 1, "attests")?, "attests")? {
83911×
            b.attests.push(attest_from(a)?);
840
        }
841112×
        if let Some(ps) = m.get(&3) {
842112×
            for p in o::array(ps, "policies")? {
8433×
                b.policies.push(policy_from(p)?);
844
            }
8450×
        }
846112×
        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
            }
853111×
        }
854112×
        Ok(b)
855112×
    }
856
85783×
    pub fn encode_bundle(b: &Bundle) -> Vec<u8> {
85883×
        cbor::encode(&bundle_value(b))
85983×
    }
86077×
    pub fn decode_bundle(bytes: &[u8]) -> Result<Bundle> {
86177×
        bundle_from_value(&cbor::decode(bytes).map_err(ObjectError::from)?)
86277×
    }
863
864123×
    pub fn encode_capabilities(id: [u8; 16]) -> Vec<u8> {
865123×
        cbor::encode(&Value::map([
866123×
            (0, Value::Array(vec![Value::Uint(PROTOCOL_VERSION)])),
867123×
            (1, Value::Bytes(id.to_vec())),
868123×
            (2, Value::Array(vec![Value::Text("seal".into())])),
869123×
        ]))
870123×
    }
871123×
    pub fn decode_capabilities(bytes: &[u8]) -> Result<[u8; 16]> {
872123×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
873123×
        let m = map(&v, "capabilities")?;
874123×
        let versions = o::array(field(m, 0, "versions")?, "versions")?;
875123×
        if !versions
876123×
            .iter()
877123×
            .any(|x| x.as_uint() == Some(PROTOCOL_VERSION))
878
        {
8790×
            return Err(RepoError::Msg(
8800×
                "remote does not speak protocol version 1".into(),
8810×
            ));
882123×
        }
883123×
        o::bytes(field(m, 1, "repo id")?, "repo id")?
884123×
            .try_into()
885123×
            .map_err(|_| schema("repo id: expected 16 bytes"))
886123×
    }
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
916147×
    pub fn encode_refs(id: [u8; 16], resp: &RefsResponse) -> Vec<u8> {
917147×
        cbor::encode(&Value::map([
918
            (
919
                0,
920
                Value::Array(
921147×
                    resp.refs
922147×
                        .iter()
923151×
                        .map(|r| {
924151×
                            let mut v = Value::map([
925151×
                                (0, Value::Text(r.name.clone())),
926151×
                                (1, Value::Bytes(r.state.0.to_vec())),
927151×
                                (2, hashes_value(&r.heads)),
928151×
                                (3, opt_attest_value(&r.attestation)),
929151×
                                (4, r.last_serial.map(Value::Uint).unwrap_or(Value::Null)),
930151×
                            ]);
931
                            // Only when there is one; older peers ignore it.
932151×
                            if r.accepted.is_some() {
9330×
                                if let Value::Map(m) = &mut v {
9340×
                                    m.insert(5, opt_attest_value(&r.accepted));
9350×
                                }
936151×
                            }
937151×
                            v
938151×
                        })
939147×
                        .collect(),
940
                ),
941
            ),
942147×
            (1, Value::Bytes(id.to_vec())),
943147×
            (2, opt_policy_value(resp.policy.as_ref())),
944147×
            (3, tags_value(&resp.tags)),
945147×
            (
946147×
                4,
947147×
                Value::Array(resp.history.iter().map(policy_value).collect()),
948147×
            ),
949147×
            (
950147×
                5,
951147×
                resp.default_bookmark
952147×
                    .clone()
953147×
                    .map(Value::Text)
954147×
                    .unwrap_or(Value::Null),
955147×
            ),
956
        ]))
957147×
    }
958190×
    pub fn decode_refs(bytes: &[u8]) -> Result<([u8; 16], RefsResponse)> {
959190×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
960190×
        let m = map(&v, "refs")?;
961190×
        let id: [u8; 16] = o::bytes(field(m, 1, "repo id")?, "repo id")?
962190×
            .try_into()
963190×
            .map_err(|_| schema("repo id: expected 16 bytes"))?;
964190×
        let mut refs = Vec::new();
965194×
        for r in o::array(field(m, 0, "refs")?, "refs")? {
966194×
            let rm = map(r, "ref")?;
967194×
            let ls = field(rm, 4, "last serial")?;
968194×
            refs.push(RemoteRef {
969194×
                name: o::text(field(rm, 0, "name")?, "name")?.to_owned(),
970194×
                state: o::oref(field(rm, 1, "state")?, "state")?,
971194×
                heads: hashes_from(field(rm, 2, "heads")?)?,
972194×
                attestation: opt_attest_from(field(rm, 3, "attestation")?)?,
973194×
                last_serial: if ls.is_null() {
974178×
                    None
975
                } else {
97616×
                    Some(o::uint(ls, "last serial")?)
977
                },
978194×
                accepted: match rm.get(&5) {
9790×
                    Some(a) => opt_attest_from(a)?,
980194×
                    None => None,
981
                },
982
            });
983
        }
984190×
        let policy = match m.get(&2) {
985190×
            Some(p) => opt_policy_from(p)?,
9860×
            None => None,
987
        };
988190×
        let tags = match m.get(&3) {
989190×
            Some(t) => tags_from(t)?,
9900×
            None => Vec::new(),
991
        };
992190×
        let mut history = Vec::new();
993190×
        if let Some(h) = m.get(&4) {
994190×
            for p in o::array(h, "history")? {
9951×
                history.push(policy_from(p)?);
996
            }
9970×
        }
998190×
        let default_bookmark = match m.get(&5) {
999121×
            Some(Value::Text(t)) => Some(t.clone()),
100069×
            _ => None,
1001
        };
1002190×
        Ok((
1003190×
            id,
1004190×
            RefsResponse {
1005190×
                refs,
1006190×
                policy,
1007190×
                tags,
1008190×
                history,
1009190×
                default_bookmark,
1010190×
            },
1011190×
        ))
1012190×
    }
1013
101448×
    pub fn encode_negotiate(have: &[ChangeHash], want: &[ChangeHash]) -> Vec<u8> {
101548×
        cbor::encode(&Value::map([
101648×
            (0, hashes_value(have)),
101748×
            (1, hashes_value(want)),
101848×
        ]))
101948×
    }
102048×
    pub fn decode_negotiate(bytes: &[u8]) -> Result<(Vec<ChangeHash>, Vec<ChangeHash>)> {
102148×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
102248×
        let m = map(&v, "negotiate")?;
1023
        Ok((
102448×
            hashes_from(field(m, 0, "have")?)?,
102548×
            hashes_from(field(m, 1, "want")?)?,
1026
        ))
102748×
    }
1028
1029
    /// `fetch` request: `{0: [changes], 1: max blob size or null}`.
103049×
    pub fn encode_fetch(hs: &[ChangeHash], max_blob: Option<u64>) -> Vec<u8> {
103149×
        cbor::encode(&Value::map([
103249×
            (0, hashes_value(hs)),
103349×
            (1, max_blob.map(Value::Uint).unwrap_or(Value::Null)),
103449×
        ]))
103549×
    }
103650×
    pub fn decode_fetch(bytes: &[u8]) -> Result<(Vec<ChangeHash>, Option<u64>)> {
103750×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
103850×
        let m = map(&v, "fetch")?;
103950×
        let max = match m.get(&1) {
104049×
            Some(x) if !x.is_null() => Some(o::uint(x, "max blob")?),
104148×
            _ => None,
1042
        };
104350×
        Ok((hashes_from(field(m, 0, "changes")?)?, max))
104450×
    }
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
106250×
    pub fn encode_hashes(hs: &[ChangeHash]) -> Vec<u8> {
106350×
        cbor::encode(&Value::map([(0, hashes_value(hs))]))
106450×
    }
106549×
    pub fn decode_hashes(bytes: &[u8]) -> Result<Vec<ChangeHash>> {
106649×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
106749×
        hashes_from(field(map(&v, "hashes")?, 0, "hashes")?)
106849×
    }
1069
107039×
    pub fn encode_push(
107139×
        bundle: &Bundle,
107239×
        updates: &[RefUpdate],
107339×
        policy: Option<&SignedPolicy>,
107439×
        tags: &[TagRef],
107539×
    ) -> Vec<u8> {
107639×
        cbor::encode(&Value::map([
107739×
            (0, bundle_value(bundle)),
107839×
            (2, opt_policy_value(policy)),
107939×
            (3, tags_value(tags)),
1080
            (
1081
                1,
1082
                Value::Array(
108339×
                    updates
108439×
                        .iter()
108539×
                        .map(|u| {
108639×
                            Value::map([
108739×
                                (0, Value::Text(u.name.clone())),
108839×
                                (1, Value::Bytes(u.old_state.0.to_vec())),
108939×
                                (2, Value::Bytes(u.new_state.0.to_vec())),
109039×
                                (3, Value::Bool(u.force)),
109139×
                                (4, opt_attest_value(&u.attestation)),
109239×
                            ])
109339×
                        })
109439×
                        .collect(),
1095
                ),
1096
            ),
1097
        ]))
109839×
    }
1099
    /// A decoded push: bundle, bookmark updates, policy, tags.
1100
    pub type PushParts = (Bundle, Vec<RefUpdate>, Option<SignedPolicy>, Vec<TagRef>);
1101
110235×
    pub fn decode_push(bytes: &[u8]) -> Result<PushParts> {
110335×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
110435×
        let m = map(&v, "push")?;
110535×
        let bundle = bundle_from_value(field(m, 0, "bundle")?)?;
110635×
        let mut updates = Vec::new();
110735×
        for u in o::array(field(m, 1, "updates")?, "updates")? {
110835×
            let um = map(u, "update")?;
110935×
            updates.push(RefUpdate {
111035×
                name: o::text(field(um, 0, "name")?, "name")?.to_owned(),
111135×
                old_state: o::oref(field(um, 1, "old_state")?, "old_state")?,
111235×
                new_state: o::oref(field(um, 2, "new_state")?, "new_state")?,
111335×
                force: field(um, 3, "force")?
111435×
                    .as_bool()
111535×
                    .ok_or_else(|| schema("force: expected bool"))?,
111635×
                attestation: opt_attest_from(field(um, 4, "attestation")?)?,
1117
            });
1118
        }
111935×
        let policy = match m.get(&2) {
112035×
            Some(p) => opt_policy_from(p)?,
11210×
            None => None,
1122
        };
112335×
        let tags = match m.get(&3) {
112435×
            Some(t) => tags_from(t)?,
11250×
            None => Vec::new(),
1126
        };
112735×
        Ok((bundle, updates, policy, tags))
112835×
    }
1129
113035×
    pub fn encode_push_response(results: &[RefResult]) -> Vec<u8> {
113135×
        cbor::encode(&Value::map([(
1132
            0,
1133
            Value::Array(
113435×
                results
113535×
                    .iter()
113636×
                    .map(|r| {
113736×
                        Value::map([
113836×
                            (0, Value::Text(r.name.clone())),
113936×
                            (1, Value::Bool(r.ok)),
114036×
                            (2, Value::Text(r.reason.clone())),
114136×
                        ])
114236×
                    })
114335×
                    .collect(),
1144
            ),
1145
        )]))
114635×
    }
114735×
    pub fn decode_push_response(bytes: &[u8]) -> Result<Vec<RefResult>> {
114835×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
114935×
        let m = map(&v, "push response")?;
115035×
        let mut out = Vec::new();
115136×
        for r in o::array(field(m, 0, "results")?, "results")? {
115236×
            let rm = map(r, "result")?;
115336×
            out.push(RefResult {
115436×
                name: o::text(field(rm, 0, "name")?, "name")?.to_owned(),
115536×
                ok: field(rm, 1, "ok")?
115636×
                    .as_bool()
115736×
                    .ok_or_else(|| schema("ok: expected bool"))?,
115836×
                reason: o::text(field(rm, 2, "reason")?, "reason")?.to_owned(),
1159
            });
1160
        }
116135×
        Ok(out)
116235×
    }
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
}