OXYGENHUB

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

803 of 961 lines run, 143 of 219 functions · run #312 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
}
83
84
/// Everything needed to add a set of changes to a repository.
85
#[derive(Clone, Debug, Default)]
86
pub struct Bundle {
87
    /// In dependency order, each with its individual signatures.
88
    pub changes: Vec<(Change, Vec<Signature>)>,
89
    /// Seals covering the changes.
90
    pub attests: Vec<SignedAttest>,
91
    /// Chunks, blobs, identities, and state objects referenced by the changes.
92
    pub objects: Vec<(ObjectType, Vec<u8>)>,
93
    /// Policies between the receiver's and the one sent with the push,
94
    /// ascending by serial (spec §14.5).
95
    pub policies: Vec<SignedPolicy>,
96
    /// Each change's hash, in the order of `changes`, when whoever made
97
    /// the bundle knows it: the store it came from, or the bytes it was
98
    /// decoded from. Never read from the wire. Empty when unknown.
99
    pub hashes: Vec<ChangeHash>,
100
    /// Changes as stored, with their signatures, in dependency order, for a
101
    /// fetch between repositories on one machine: their text blobs and
102
    /// chunks are among `objects`. Never on the wire.
103
    pub raw_changes: Vec<(Vec<u8>, Vec<Signature>)>,
104
    /// Certificates of short-lived keys that signed what the bundle carries
105
    /// (spec §14.7), their issuers' identities among `objects`.
106
    pub certificates: Vec<crate::cert::SignedCertificate>,
107
}
108
109
impl Bundle {
11027×
    pub fn is_empty(&self) -> bool {
11127×
        self.changes.is_empty() && self.attests.is_empty() && self.objects.is_empty()
11227×
    }
113
}
114
115
/// A requested bookmark move (spec §10.2, §16.3).
116
#[derive(Clone, Debug)]
117
pub struct RefUpdate {
118
    pub name: String,
119
    pub old_state: ObjectRef,
120
    pub new_state: ObjectRef,
121
    pub force: bool,
122
    pub attestation: Option<SignedAttest>,
123
}
124
125
#[derive(Clone, Debug, PartialEq, Eq)]
126
pub struct RefResult {
127
    pub name: String,
128
    pub ok: bool,
129
    pub reason: String,
130
}
131
132
pub trait Remote {
133
    fn repo_id(&self) -> Result<[u8; 16]>;
134
    fn refs(&self) -> Result<RefsResponse>;
135
    /// Changes in the closure of `want` but not of `have`, in dependency order.
136
    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>>;
137
    /// The changes with their signatures, seals, identities, and blobs; blobs
138
    /// larger than `max_blob` bytes are left out for lazy fetching (spec §13.5).
139
    fn fetch(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle>;
140
    /// Individual objects by reference (spec §16.1 `objects`); a blob comes
141
    /// with its chunks.
142
    fn objects(&self, refs: &[ObjectRef]) -> Result<Vec<(ObjectType, Vec<u8>)>>;
143
    /// A checkpoint of exactly `heads` (spec §16.1 `checkpoint`): the
144
    /// checkpoint object, its payload blob, and the blob's chunks.
145
    fn checkpoint(&self, heads: &[ChangeHash]) -> Result<Bundle>;
146
    /// The changes within `depth` dependency steps of `heads` (spec §17).
147
    fn shallow(&self, heads: &[ChangeHash], depth: u32) -> Result<Vec<ChangeHash>>;
148
    /// Push changes, bookmark updates, new tags, and optionally a newer
149
    /// policy. The policy is applied first; a result named `policy` reports
150
    /// it, and each tag reports as `tags/<name>`.
151
    fn push(
152
        &self,
153
        bundle: Bundle,
154
        updates: Vec<RefUpdate>,
155
        policy: Option<SignedPolicy>,
156
        tags: Vec<TagRef>,
157
    ) -> Result<Vec<RefResult>>;
158
}
159
160
// ---------------------------------------------------------------- local
161
162
/// Another repository on this machine, served in-process.
163
pub struct LocalRemote {
164
    pub repo: Repo,
165
    /// The pushing repository's id, which binds an unbound target on the
166
    /// first push, as `Oxy-Repo-Id` does over HTTP.
167
    pub claimed_id: Option<[u8; 16]>,
168
}
169
170
impl LocalRemote {
1711046×
    pub fn open(path: &Path) -> Result<LocalRemote> {
172
        Ok(LocalRemote {
1731046×
            repo: Repo::open(path)?,
1741046×
            claimed_id: None,
175
        })
1761046×
    }
177
}
178
179
impl Remote for LocalRemote {
180823×
    fn repo_id(&self) -> Result<[u8; 16]> {
181823×
        self.repo.id()
182823×
    }
183822×
    fn refs(&self) -> Result<RefsResponse> {
184822×
        self.repo.remote_refs()
185822×
    }
186308×
    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>> {
187308×
        self.repo.negotiate(have, want)
188308×
    }
189311×
    fn fetch(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle> {
190311×
        self.repo.bundle_raw_for(changes, max_blob)
191311×
    }
1923×
    fn objects(&self, refs: &[ObjectRef]) -> Result<Vec<(ObjectType, Vec<u8>)>> {
1933×
        self.repo.objects_for(refs)
1943×
    }
1951×
    fn checkpoint(&self, heads: &[ChangeHash]) -> Result<Bundle> {
1961×
        self.repo.checkpoint_bundle_for(heads)
1971×
    }
1983×
    fn shallow(&self, heads: &[ChangeHash], depth: u32) -> Result<Vec<ChangeHash>> {
1993×
        self.repo.shallow_changes(heads, depth)
2003×
    }
201244×
    fn push(
202244×
        &self,
203244×
        bundle: Bundle,
204244×
        updates: Vec<RefUpdate>,
205244×
        policy: Option<SignedPolicy>,
206244×
        tags: Vec<TagRef>,
207244×
    ) -> Result<Vec<RefResult>> {
208244×
        if self.repo.is_unbound() {
20911×
            match self.claimed_id {
21011×
                Some(id) => self.repo.bind_id(id)?,
211
                None => {
2120×
                    return Err(RepoError::Msg(
2130×
                        "the remote is unbound and this client did not say which repository it is"
2140×
                            .into(),
2150×
                    ))
216
                }
217
            }
218233×
        }
219244×
        self.repo.receive_push(bundle, updates, policy, tags)
220244×
    }
221
}
222
223
// ---------------------------------------------------------------- http
224
225
/// How to reach a remote: credentials and trust.
226
#[derive(Clone, Debug, Default)]
227
pub struct RemoteOptions {
228
    /// Bearer token for pushes.
229
    pub token: Option<String>,
230
    /// PEM file with extra root certificates to trust for `https://` remotes
231
    /// (self-hosted servers); the system web roots are always trusted.
232
    pub ca: Option<PathBuf>,
233
    /// The local repository's id, sent as `Oxy-Repo-Id` so an unbound
234
    /// hosted repository can adopt it on the first push (spec §16.1).
235
    pub repo_id: Option<[u8; 16]>,
236
    /// Where a remote kept on a store (`dir://`, `s3://`, …) keeps its
237
    /// mirror and what it has seen: the remote's directory under
238
    /// `.oxy/remotes`. None (a clone, before there is a repository) means
239
    /// a temporary directory.
240
    pub cache_dir: Option<PathBuf>,
241
    /// Other tokens for the same hub, tried in turn when `token` is
242
    /// refused (401, 403 or 404): those of enclosing checkouts, when
243
    /// `token` came from the credential store rather than from this
244
    /// remote's own configuration (oxygen#112).
245
    pub fallback_tokens: Vec<String>,
246
}
247
248
/// Opens a store for a URL it understands, or passes (`None`). Stores that
249
/// need more than the standard library (the cloud ones) are added by the
250
/// program that links them, with [`register_store_opener`].
251
pub type StoreOpener =
252
    fn(&str, &RemoteOptions) -> Option<crate::strata::Result<Box<dyn crate::strata::Store>>>;
253
254
static STORE_OPENERS: Mutex<Vec<StoreOpener>> = Mutex::new(Vec::new());
255
256154×
pub fn register_store_opener(opener: StoreOpener) {
257154×
    STORE_OPENERS
258154×
        .lock()
259154×
        .unwrap_or_else(|p| p.into_inner())
260154×
        .push(opener);
261154×
}
262
263
/// A store for `url`, from `dir://` or any registered opener.
2641106×
pub fn open_store(
2651106×
    url: &str,
2661106×
    options: &RemoteOptions,
2671106×
) -> Option<crate::strata::Result<Box<dyn crate::strata::Store>>> {
2681106×
    if let Some(path) = url.strip_prefix("dir://") {
269
        return Some(
27059×
            crate::strata::LocalStore::open(path)
27159×
                .map(|s| Box::new(s) as Box<dyn crate::strata::Store>),
272
        );
2731047×
    }
2741047×
    let openers = STORE_OPENERS.lock().unwrap_or_else(|p| p.into_inner());
2751047×
    openers.iter().find_map(|open| open(url, options))
2761106×
}
277
278
/// The request header carrying the client's repository id.
279
pub const REPO_ID_HEADER: &str = "Oxy-Repo-Id";
280
281
/// The id of a hosted repository that has not adopted one yet.
282
pub const UNBOUND_ID: [u8; 16] = [0; 16];
283
284
/// A remote served by `oxy-server` at `<base>/oxy/v1/...`.
285
pub struct HttpRemote {
286
    base: String,
287
    /// The token in use, and those still to try when it is refused.
288
    token: Mutex<Option<String>>,
289
    fallbacks: Mutex<Vec<String>>,
290
    repo_id: Option<[u8; 16]>,
291
    agent: ureq::Agent,
292
}
293
294
/// A TLS client configuration trusting the web roots plus `ca`, if given:
295
/// the trust every oxy connection uses, shared with tools that call hub APIs.
296153×
pub fn tls_client_config(ca: Option<&Path>) -> Result<Arc<rustls::ClientConfig>> {
297153×
    let mut roots = rustls::RootCertStore::empty();
298153×
    roots.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
299153×
    if let Some(path) = ca {
30067×
        let pem = std::fs::read(path)
30167×
            .map_err(|e| RepoError::Msg(format!("reading CA file {}: {e}", path.display())))?;
30267×
        let certs = rustls_pemfile::certs(&mut pem.as_slice())
30367×
            .collect::<std::result::Result<Vec<_>, _>>()
30467×
            .map_err(|e| RepoError::Msg(format!("parsing CA file {}: {e}", path.display())))?;
30567×
        if certs.is_empty() {
3060×
            return Err(RepoError::Msg(format!(
3070×
                "CA file {} holds no certificates",
3080×
                path.display()
3090×
            )));
31067×
        }
31167×
        for c in certs {
31267×
            roots
31367×
                .add(c)
31467×
                .map_err(|e| RepoError::Msg(format!("CA file {}: {e}", path.display())))?;
315
        }
31686×
    }
317153×
    let cfg = rustls::ClientConfig::builder_with_provider(Arc::new(
318153×
        rustls::crypto::ring::default_provider(),
319
    ))
320153×
    .with_safe_default_protocol_versions()
321153×
    .map_err(|e| RepoError::Msg(e.to_string()))?
322153×
    .with_root_certificates(roots)
323153×
    .with_no_client_auth();
324153×
    Ok(Arc::new(cfg))
325153×
}
326
327
impl HttpRemote {
328149×
    pub fn new(base: &str, options: &RemoteOptions) -> Result<HttpRemote> {
329149×
        refuse_plain_token(base, options)?;
330149×
        let agent = ureq::AgentBuilder::new()
331149×
            .tls_config(tls_client_config(options.ca.as_deref())?)
332149×
            .build();
333149×
        Ok(HttpRemote {
334149×
            base: base.trim_end_matches('/').to_string(),
335149×
            token: Mutex::new(options.token.clone()),
336149×
            fallbacks: Mutex::new(options.fallback_tokens.iter().rev().cloned().collect()),
337149×
            repo_id: options.repo_id,
338149×
            agent,
339149×
        })
340149×
    }
341
342
    /// A request, tried again with the next fallback token while the
343
    /// hub refuses the one in use (oxygen#112).
344374×
    fn call(&self, post: bool, endpoint: &str, body: Option<&[u8]>) -> Result<Vec<u8>> {
345
        loop {
346375×
            let token = self.token.lock().unwrap_or_else(|p| p.into_inner()).clone();
347375×
            match self.call_once(post, endpoint, body, token.as_deref()) {
34811×
                Err(RepoError::Msg(m)) if token.is_some() && refused(&m) => {
3493×
                    let next = self
3503×
                        .fallbacks
3513×
                        .lock()
3523×
                        .unwrap_or_else(|p| p.into_inner())
3533×
                        .pop();
3543×
                    let Some(next) = next else {
3552×
                        return Err(RepoError::Msg(m));
356
                    };
3571×
                    crate::credentials::warn_once(&format!(
3581×
                        "{}: the stored token was refused; using the one an enclosing checkout has for this hub",
3591×
                        self.base
3601×
                    ));
3611×
                    *self.token.lock().unwrap_or_else(|p| p.into_inner()) = Some(next);
362
                }
363372×
                other => return other,
364
            }
365
        }
366374×
    }
367
368375×
    fn call_once(
369375×
        &self,
370375×
        post: bool,
371375×
        endpoint: &str,
372375×
        body: Option<&[u8]>,
373375×
        token: Option<&str>,
374375×
    ) -> Result<Vec<u8>> {
375375×
        let url = format!("{}/oxy/v1/{endpoint}", self.base);
376375×
        let mut req = if post {
377128×
            self.agent.post(&url)
378
        } else {
379247×
            self.agent.get(&url)
380
        };
381375×
        req = req.set("Accept", CONTENT_TYPE);
382375×
        if let Some(t) = token {
38322×
            req = req.set("Authorization", &format!("Bearer {t}"));
384353×
        }
385375×
        if let Some(id) = &self.repo_id {
386350×
            req = req.set(REPO_ID_HEADER, &hex::encode(id));
387350×
        }
388375×
        let resp = match body {
389128×
            Some(b) => req.set("Content-Type", CONTENT_TYPE).send_bytes(b),
390247×
            None => req.call(),
391
        };
39211×
        match resp {
393364×
            Ok(r) => {
394364×
                let mut out = Vec::new();
395364×
                r.into_reader()
396364×
                    .read_to_end(&mut out)
397364×
                    .map_err(|e| RepoError::Msg(format!("{url}: {e}")))?;
398364×
                Ok(out)
399
            }
4008×
            Err(ureq::Error::Status(code, r)) => {
4018×
                let text = r.into_string().unwrap_or_default();
4028×
                let msg = format!("{url}: HTTP {code}: {}", text.trim());
4038×
                Err(RepoError::Msg(with_hint(msg)))
404
            }
405
            // The transport error already names the URL.
4063×
            Err(e) => Err(RepoError::Msg(with_hint(e.to_string()))),
407
        }
408375×
    }
409
}
410
411
/// Whether a failed request's message says the hub refused its token:
412
/// HTTP 401, 403, or 404 (a hub hides a private repository from a token
413
/// that may not read it).
4144×
fn refused(message: &str) -> bool {
4154×
    [": HTTP 401", ": HTTP 403", ": HTTP 404"]
4164×
        .iter()
4176×
        .any(|c| message.contains(c))
4184×
}
419
420
/// Refuse to send a token where anyone on the way could read it: plain
421
/// `http` to another machine (see [`crate::credentials::token_allowed`]).
422149×
fn refuse_plain_token(url: &str, options: &RemoteOptions) -> Result<()> {
423149×
    if options.token.is_some() && !crate::credentials::token_allowed(url) {
4240×
        return Err(RepoError::Msg(format!(
4250×
            "{url}: refusing to send a token over plain http; use https"
4260×
        )));
427149×
    }
428149×
    Ok(())
429149×
}
430
431
/// What to do about a failed request, recognised from its message.
43219×
pub fn hint_for(message: &str) -> Option<&'static str> {
43319×
    let m = message.to_ascii_lowercase();
43419×
    let hint = if m.contains("http 401") {
4356×
        "the hub needs you signed in, or did not accept the token: run `oxy auth login HUB` (pushing needs repo:write)"
43613×
    } else if m.contains("http 403") {
4371×
        "the token or account lacks permission: pushing needs a token with repo:write and write access to the repository"
43812×
    } else if m.contains("http 404") {
4394×
        "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`)"
4408×
    } else if m.contains("unknownissuer")
4416×
        || m.contains("unknown issuer")
4426×
        || m.contains("certificate")
4435×
        || m.contains("invalidcertificate")
444
    {
4453×
        "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`"
4465×
    } else if m.contains("connection refused") {
4473×
        "nothing is listening there: check the URL and port, and that the server is running"
4482×
    } else if m.contains("dns")
4491×
        || m.contains("failed to lookup")
4501×
        || m.contains("name or service not known")
451
    {
4521×
        "the host name does not resolve: check the URL"
4531×
    } else if m.contains("timed out") || m.contains("timeout") {
4540×
        "the server did not answer in time: check the network, or try again"
455
    } else {
4561×
        return None;
457
    };
45818×
    Some(hint)
45919×
}
460
461
/// `message`, followed by what to do about it when that is known.
46211×
pub fn with_hint(message: String) -> String {
46311×
    match hint_for(&message) {
46411×
        Some(h) => format!("{message}\n  hint: {h}"),
4650×
        None => message,
466
    }
46711×
}
468
469
/// A JSON request to a hub API (`GET`, `POST` or `DELETE`), with the same
470
/// TLS trust and token rules as remotes. Returns the status and body, an
471
/// error status included; only transport failures are errors.
4720×
pub fn http_json(
4730×
    method: &str,
4740×
    url: &str,
4750×
    body: Option<&str>,
4760×
    options: &RemoteOptions,
4770×
) -> Result<(u16, String)> {
4780×
    refuse_plain_token(url, options)?;
4790×
    let agent = ureq::AgentBuilder::new()
4800×
        .tls_config(tls_client_config(options.ca.as_deref())?)
4810×
        .timeout(std::time::Duration::from_secs(60))
4820×
        .build();
4830×
    let mut req = agent.request(method, url).set("Accept", "application/json");
4840×
    if let Some(t) = &options.token {
4850×
        req = req.set("Authorization", &format!("Bearer {t}"));
4860×
    }
4870×
    let resp = match body {
4880×
        Some(b) => req.set("Content-Type", "application/json").send_string(b),
4890×
        None => req.call(),
490
    };
4910×
    match resp {
4920×
        Ok(r) => {
4930×
            let code = r.status();
4940×
            Ok((code, r.into_string().unwrap_or_default()))
495
        }
4960×
        Err(ureq::Error::Status(code, r)) => Ok((code, r.into_string().unwrap_or_default())),
4970×
        Err(e) => Err(RepoError::Msg(with_hint(e.to_string()))),
498
    }
4990×
}
500
501
/// An authenticated GET of any URL on a host, with the same TLS trust and
502
/// token as remotes use (for hub APIs: proposals). Returns the body.
5030×
pub fn http_get(url: &str, options: &RemoteOptions) -> Result<Vec<u8>> {
5040×
    refuse_plain_token(url, options)?;
5050×
    let agent = ureq::AgentBuilder::new()
5060×
        .tls_config(tls_client_config(options.ca.as_deref())?)
5070×
        .build();
5080×
    let mut req = agent.get(url).set("Accept", "application/json");
5090×
    if let Some(t) = &options.token {
5100×
        req = req.set("Authorization", &format!("Bearer {t}"));
5110×
    }
5120×
    match req.call() {
5130×
        Ok(r) => {
5140×
            let mut out = Vec::new();
5150×
            r.into_reader()
5160×
                .read_to_end(&mut out)
5170×
                .map_err(|e| RepoError::Msg(format!("{url}: {e}")))?;
5180×
            Ok(out)
519
        }
5200×
        Err(ureq::Error::Status(code, r)) => {
5210×
            let text = r.into_string().unwrap_or_default();
5220×
            Err(RepoError::Msg(with_hint(format!(
5230×
                "{url}: HTTP {code}: {}",
5240×
                text.trim()
5250×
            ))))
526
        }
527
        // The transport error already names the URL.
5280×
        Err(e) => Err(RepoError::Msg(with_hint(e.to_string()))),
529
    }
5300×
}
531
532
impl Remote for HttpRemote {
533128×
    fn repo_id(&self) -> Result<[u8; 16]> {
534128×
        wire::decode_capabilities(&self.call(false, "capabilities", None)?)
535128×
    }
536119×
    fn refs(&self) -> Result<RefsResponse> {
537119×
        Ok(wire::decode_refs(&self.call(false, "refs", None)?)?.1)
538119×
    }
53947×
    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>> {
54047×
        wire::decode_hashes(&self.call(
541
            true,
54247×
            "negotiate",
54347×
            Some(&wire::encode_negotiate(have, want)),
5440×
        )?)
54547×
    }
54647×
    fn fetch(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle> {
54747×
        wire::decode_bundle(&self.call(
548
            true,
54947×
            "fetch",
55047×
            Some(&wire::encode_fetch(changes, max_blob)),
5510×
        )?)
55247×
    }
5531×
    fn objects(&self, refs: &[ObjectRef]) -> Result<Vec<(ObjectType, Vec<u8>)>> {
554
        Ok(
5551×
            wire::decode_bundle(&self.call(true, "objects", Some(&wire::encode_hashes(refs)))?)?
556
                .objects,
557
        )
5581×
    }
5590×
    fn checkpoint(&self, heads: &[ChangeHash]) -> Result<Bundle> {
5600×
        wire::decode_bundle(&self.call(true, "checkpoint", Some(&wire::encode_hashes(heads)))?)
5610×
    }
5620×
    fn shallow(&self, heads: &[ChangeHash], depth: u32) -> Result<Vec<ChangeHash>> {
5630×
        wire::decode_hashes(&self.call(
564
            true,
5650×
            "shallow",
5660×
            Some(&wire::encode_shallow(heads, depth)),
5670×
        )?)
5680×
    }
56932×
    fn push(
57032×
        &self,
57132×
        bundle: Bundle,
57232×
        updates: Vec<RefUpdate>,
57332×
        policy: Option<SignedPolicy>,
57432×
        tags: Vec<TagRef>,
57532×
    ) -> Result<Vec<RefResult>> {
57632×
        wire::decode_push_response(&self.call(
577
            true,
57832×
            "push",
57932×
            Some(&wire::encode_push(
58032×
                &bundle,
58132×
                &updates,
58232×
                policy.as_ref(),
58332×
                &tags,
58432×
            )),
5854×
        )?)
58632×
    }
587
}
588
589
/// A remote's URL as it is stored: an `http(s)://` URL as given, a local
590
/// path made absolute (with its `file://` kept), so it still names the same
591
/// repository from inside the clone (oxygen#57).
592254×
pub fn stored_url(url: &str) -> String {
593
    // Only a path is made absolute: a plain one, or one behind file:// or
594
    // dir://. Every other URL (a hub, a bucket) is kept as written.
595254×
    let (prefix, path) = match (url.strip_prefix("file://"), url.strip_prefix("dir://")) {
5961×
        (Some(p), _) => ("file://", p),
59711×
        (_, Some(p)) => ("dir://", p),
598242×
        _ if url.contains("://") => return url.to_string(),
599219×
        _ => ("", url),
600
    };
601231×
    match std::path::absolute(path) {
602231×
        Ok(abs) => format!("{prefix}{}", abs.display()),
6030×
        Err(_) => url.to_string(),
604
    }
605254×
}
606
607
/// Open a remote by URL: a local path (optionally `file://`), `http(s)://`
608
/// (a hub), or a store (`dir://`, and whatever openers are registered).
6091243×
pub fn open_remote(url: &str, options: &RemoteOptions) -> Result<Box<dyn Remote>> {
6101243×
    if url.starts_with("http://") || url.starts_with("https://") {
611149×
        return Ok(Box::new(HttpRemote::new(url, options)?));
6121094×
    }
6131094×
    if let Some(store) = open_store(url, options) {
61448×
        return crate::repo::strata::open_store_remote(store?, options);
6151046×
    }
6161046×
    let path = url.strip_prefix("file://").unwrap_or(url);
6171046×
    if path.contains("://") {
6180×
        return Err(RepoError::Msg(format!("unsupported remote URL {url:?}")));
6191046×
    }
6201046×
    let mut local = LocalRemote::open(Path::new(path))?;
6211046×
    local.claimed_id = options.repo_id;
6221046×
    Ok(Box::new(local))
6231243×
}
624
625
// ---------------------------------------------------------------- wire
626
627
/// CBOR encodings of the protocol messages (spec §16.1).
628
pub mod wire {
629
    use super::*;
630
6310×
    fn schema(msg: &str) -> RepoError {
6320×
        RepoError::Object(ObjectError::Schema(msg.into()))
6330×
    }
634145×
    fn sigs_value(sigs: &[Signature]) -> Value {
635145×
        Value::Array(sigs.iter().map(|s| s.to_value()).collect())
636145×
    }
637140×
    fn sigs_from(v: &Value) -> Result<Vec<Signature>> {
638140×
        Ok(o::array(v, "signatures")?
639140×
            .iter()
640140×
            .map(Signature::from_value)
641140×
            .collect::<std::result::Result<Vec<_>, _>>()?)
642140×
    }
643354×
    fn hashes_value(hs: &[ObjectRef]) -> Value {
644496×
        Value::Array(hs.iter().map(|h| Value::Bytes(h.0.to_vec())).collect())
645354×
    }
646413×
    fn hashes_from(v: &Value) -> Result<Vec<ObjectRef>> {
647413×
        Ok(o::array(v, "hashes")?
648413×
            .iter()
649556×
            .map(|h| o::oref(h, "hash"))
650413×
            .collect::<std::result::Result<Vec<_>, _>>()?)
651413×
    }
65224×
    fn attest_value(sa: &SignedAttest) -> Value {
65324×
        Value::Array(vec![Value::Bytes(sa.attest.encode()), sigs_value(&sa.sigs)])
65424×
    }
65538×
    fn attest_from(v: &Value) -> Result<SignedAttest> {
65638×
        let a = o::array(v, "signed attest")?;
65738×
        if a.len() != 2 {
6580×
            return Err(schema("signed attest: expected [attest, sigs]"));
65938×
        }
660
        Ok(SignedAttest {
66138×
            attest: Attest::decode(o::bytes(&a[0], "attest")?)?,
66238×
            sigs: sigs_from(&a[1])?,
663
        })
66438×
    }
665191×
    fn opt_attest_value(sa: &Option<SignedAttest>) -> Value {
666191×
        sa.as_ref().map(attest_value).unwrap_or(Value::Null)
667191×
    }
668246×
    fn opt_attest_from(v: &Value) -> Result<Option<SignedAttest>> {
669246×
        if v.is_null() {
670219×
            Ok(None)
671
        } else {
67227×
            attest_from(v).map(Some)
673
        }
674246×
    }
675
    /// `[policy bytes, [signature]]`: also the format of a policy proposal
676
    /// file exchanged between root holders (spec §14.5).
67710×
    pub fn policy_value(sp: &SignedPolicy) -> Value {
67810×
        Value::Array(vec![Value::Bytes(sp.policy.encode()), sigs_value(&sp.sigs)])
67910×
    }
6808×
    pub fn policy_from(v: &Value) -> Result<SignedPolicy> {
6818×
        let a = o::array(v, "signed policy")?;
6828×
        if a.len() != 2 {
6830×
            return Err(schema("signed policy: expected [policy, sigs]"));
6848×
        }
685
        Ok(SignedPolicy {
6868×
            policy: Policy::decode(o::bytes(&a[0], "policy")?)?,
6878×
            sigs: sigs_from(&a[1])?,
688
        })
6898×
    }
690180×
    fn opt_policy_value(sp: Option<&SignedPolicy>) -> Value {
691180×
        sp.map(policy_value).unwrap_or(Value::Null)
692180×
    }
693219×
    fn opt_policy_from(v: &Value) -> Result<Option<SignedPolicy>> {
694219×
        if v.is_null() {
695216×
            Ok(None)
696
        } else {
6973×
            policy_from(v).map(Some)
698
        }
699219×
    }
700901×
    fn map<'a>(v: &'a Value, what: &str) -> Result<&'a o::Map> {
701901×
        Ok(o::map(v, what)?)
702901×
    }
7032525×
    fn field<'a>(m: &'a o::Map, k: u64, what: &str) -> Result<&'a Value> {
7042525×
        Ok(o::field(m, k, what)?)
7052525×
    }
706
707
    /// A bundle's value. Each change's text blobs and their chunks (spec
708
    /// §6.1) are added to its objects when they are not already there, so a
709
    /// bundle can always be read on its own.
710116×
    pub fn bundle_value(b: &Bundle) -> Value {
711116×
        let mut changes = Vec::with_capacity(b.raw_changes.len() + b.changes.len());
712
        // Changes as stored are already their encoding, their texts among
713
        // the objects.
714116×
        for (bytes, s) in &b.raw_changes {
7150×
            changes.push(Value::Array(vec![
7160×
                Value::Bytes(bytes.clone()),
7170×
                sigs_value(s),
7180×
            ]));
7190×
        }
720116×
        let mut texts: Vec<(ObjectType, Vec<u8>)> = Vec::new();
721116×
        let mut listed: Option<std::collections::HashSet<ObjectRef>> = None;
722116×
        for (c, s) in &b.changes {
723111×
            let (bytes, stored) = c.encode_parts();
724111×
            for t in stored {
7250×
                let listed = listed.get_or_insert_with(|| {
7260×
                    b.objects
7270×
                        .iter()
7280×
                        .map(|(ty, body)| crate::hash::hash_object(*ty, body))
7290×
                        .collect()
7300×
                });
7310×
                if !listed.insert(t.blob.hash()) {
7320×
                    continue;
7330×
                }
7340×
                for chunk in crate::chunk::chunks(&t.content) {
7350×
                    if listed.insert(crate::hash::hash_object(ObjectType::Chunk, chunk)) {
7360×
                        texts.push((ObjectType::Chunk, chunk.to_vec()));
7370×
                    }
738
                }
7390×
                texts.push((ObjectType::Blob, t.blob.encode()));
740
            }
741111×
            changes.push(Value::Array(vec![Value::Bytes(bytes), sigs_value(s)]));
742
        }
743116×
        let mut v = Value::map([
744116×
            (0, Value::Array(changes)),
745116×
            (
746116×
                1,
747116×
                Value::Array(b.attests.iter().map(attest_value).collect()),
748116×
            ),
749
            (
750
                2,
751
                Value::Array(
752116×
                    b.objects
753116×
                        .iter()
754116×
                        .chain(&texts)
755116×
                        .map(|(t, body)| {
75686×
                            Value::Array(vec![
75786×
                                Value::Text(t.name().into()),
75886×
                                Value::Bytes(body.clone()),
75986×
                            ])
76086×
                        })
761116×
                        .collect(),
762
                ),
763
            ),
764116×
            (
765116×
                3,
766116×
                Value::Array(b.policies.iter().map(policy_value).collect()),
767116×
            ),
768
        ]);
769
        // Only when there are any; older peers ignore the field.
770116×
        if !b.certificates.is_empty() {
7711×
            if let Value::Map(m) = &mut v {
7721×
                m.insert(
773
                    4,
774
                    Value::Array(
7751×
                        b.certificates
7761×
                            .iter()
7771×
                            .map(|c| Value::Bytes(c.encode()))
7781×
                            .collect(),
779
                    ),
780
                );
7810×
            }
782115×
        }
783116×
        v
784116×
    }
785
786106×
    pub fn bundle_from_value(v: &Value) -> Result<Bundle> {
787106×
        let m = map(v, "bundle")?;
788106×
        let mut b = Bundle::default();
789106×
        for ob in o::array(field(m, 2, "objects")?, "objects")? {
79075×
            let pair = o::array(ob, "object")?;
79175×
            if pair.len() != 2 {
7920×
                return Err(schema("object: expected [type, body]"));
79375×
            }
79475×
            let ty = ObjectType::from_name(o::text(&pair[0], "object type")?)
79575×
                .ok_or_else(|| schema("unknown object type"))?;
79675×
            b.objects
79775×
                .push((ty, o::bytes(&pair[1], "object body")?.to_vec()));
798
        }
799
        // Stored inserts are read from the bundle's own objects.
800106×
        let mut by_ref: Option<std::collections::HashMap<ObjectRef, usize>> = None;
801106×
        for c in o::array(field(m, 0, "changes")?, "changes")? {
80294×
            let pair = o::array(c, "change entry")?;
80394×
            if pair.len() != 2 {
8040×
                return Err(schema("change entry: expected [change, sigs]"));
80594×
            }
80694×
            let encoded = o::bytes(&pair[0], "change")?;
80794×
            b.hashes
80894×
                .push(crate::hash::hash_object(ObjectType::Change, encoded));
80994×
            let change = Change::decode_with(encoded, &mut |text, count, form| {
8100×
                let by_ref = by_ref.get_or_insert_with(|| {
8110×
                    b.objects
8120×
                        .iter()
8130×
                        .enumerate()
8140×
                        .map(|(i, (ty, body))| (crate::hash::hash_object(*ty, body), i))
8150×
                        .collect()
8160×
                });
8170×
                let get =
8180×
                    |r: &ObjectRef, want: ObjectType| match by_ref.get(r).map(|i| &b.objects[*i]) {
8190×
                        Some((ty, body)) if *ty == want => Ok(body),
8200×
                        _ => Err(ObjectError::Schema(format!(
8210×
                            "the bundle lacks {} {}",
8220×
                            want.name(),
8230×
                            r.short(12)
8240×
                        ))),
8250×
                    };
8260×
                let blob = crate::objects::Blob::decode(get(text, ObjectType::Blob)?)?;
8270×
                let mut content = Vec::with_capacity(blob.length as usize);
8280×
                for chunk in &blob.chunks {
8290×
                    content.extend_from_slice(get(chunk, ObjectType::Chunk)?);
830
                }
8310×
                crate::objects::text_lines(&content, count, form)
8320×
            })?;
83394×
            b.changes.push((change, sigs_from(&pair[1])?));
834
        }
835106×
        for a in o::array(field(m, 1, "attests")?, "attests")? {
83611×
            b.attests.push(attest_from(a)?);
837
        }
838106×
        if let Some(ps) = m.get(&3) {
839106×
            for p in o::array(ps, "policies")? {
8403×
                b.policies.push(policy_from(p)?);
841
            }
8420×
        }
843106×
        if let Some(cs) = m.get(&4) {
8441×
            for c in o::array(cs, "certificates")? {
8451×
                b.certificates.push(
8461×
                    crate::cert::SignedCertificate::decode(o::bytes(c, "certificate")?)
8471×
                        .map_err(|e| schema(&format!("certificate: {e}")))?,
848
                );
849
            }
850105×
        }
851106×
        Ok(b)
852106×
    }
853
85483×
    pub fn encode_bundle(b: &Bundle) -> Vec<u8> {
85583×
        cbor::encode(&bundle_value(b))
85683×
    }
85777×
    pub fn decode_bundle(bytes: &[u8]) -> Result<Bundle> {
85877×
        bundle_from_value(&cbor::decode(bytes).map_err(ObjectError::from)?)
85977×
    }
860
861123×
    pub fn encode_capabilities(id: [u8; 16]) -> Vec<u8> {
862123×
        cbor::encode(&Value::map([
863123×
            (0, Value::Array(vec![Value::Uint(PROTOCOL_VERSION)])),
864123×
            (1, Value::Bytes(id.to_vec())),
865123×
            (2, Value::Array(vec![Value::Text("seal".into())])),
866123×
        ]))
867123×
    }
868123×
    pub fn decode_capabilities(bytes: &[u8]) -> Result<[u8; 16]> {
869123×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
870123×
        let m = map(&v, "capabilities")?;
871123×
        let versions = o::array(field(m, 0, "versions")?, "versions")?;
872123×
        if !versions
873123×
            .iter()
874123×
            .any(|x| x.as_uint() == Some(PROTOCOL_VERSION))
875
        {
8760×
            return Err(RepoError::Msg(
8770×
                "remote does not speak protocol version 1".into(),
8780×
            ));
879123×
        }
880123×
        o::bytes(field(m, 1, "repo id")?, "repo id")?
881123×
            .try_into()
882123×
            .map_err(|_| schema("repo id: expected 16 bytes"))
883123×
    }
884
885180×
    fn tags_value(tags: &[TagRef]) -> Value {
886
        Value::Array(
887180×
            tags.iter()
888180×
                .map(|t| {
8897×
                    Value::map([
8907×
                        (0, Value::Text(t.name.clone())),
8917×
                        (1, Value::Bytes(t.state.0.to_vec())),
8927×
                        (2, hashes_value(&t.heads)),
8937×
                        (3, opt_attest_value(&t.attestation)),
8947×
                    ])
8957×
                })
896180×
                .collect(),
897
        )
898180×
    }
899219×
    fn tags_from(v: &Value) -> Result<Vec<TagRef>> {
900219×
        let mut out = Vec::new();
901219×
        for t in o::array(v, "tags")? {
90223×
            let tm = map(t, "tag")?;
90323×
            out.push(TagRef {
90423×
                name: o::text(field(tm, 0, "name")?, "name")?.to_owned(),
90523×
                state: o::oref(field(tm, 1, "state")?, "state")?,
90623×
                heads: hashes_from(field(tm, 2, "heads")?)?,
90723×
                attestation: opt_attest_from(field(tm, 3, "attestation")?)?,
908
            });
909
        }
910219×
        Ok(out)
911219×
    }
912
913147×
    pub fn encode_refs(id: [u8; 16], resp: &RefsResponse) -> Vec<u8> {
914147×
        cbor::encode(&Value::map([
915
            (
916
                0,
917
                Value::Array(
918147×
                    resp.refs
919147×
                        .iter()
920151×
                        .map(|r| {
921151×
                            Value::map([
922151×
                                (0, Value::Text(r.name.clone())),
923151×
                                (1, Value::Bytes(r.state.0.to_vec())),
924151×
                                (2, hashes_value(&r.heads)),
925151×
                                (3, opt_attest_value(&r.attestation)),
926151×
                                (4, r.last_serial.map(Value::Uint).unwrap_or(Value::Null)),
927151×
                            ])
928151×
                        })
929147×
                        .collect(),
930
                ),
931
            ),
932147×
            (1, Value::Bytes(id.to_vec())),
933147×
            (2, opt_policy_value(resp.policy.as_ref())),
934147×
            (3, tags_value(&resp.tags)),
935147×
            (
936147×
                4,
937147×
                Value::Array(resp.history.iter().map(policy_value).collect()),
938147×
            ),
939147×
            (
940147×
                5,
941147×
                resp.default_bookmark
942147×
                    .clone()
943147×
                    .map(Value::Text)
944147×
                    .unwrap_or(Value::Null),
945147×
            ),
946
        ]))
947147×
    }
948190×
    pub fn decode_refs(bytes: &[u8]) -> Result<([u8; 16], RefsResponse)> {
949190×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
950190×
        let m = map(&v, "refs")?;
951190×
        let id: [u8; 16] = o::bytes(field(m, 1, "repo id")?, "repo id")?
952190×
            .try_into()
953190×
            .map_err(|_| schema("repo id: expected 16 bytes"))?;
954190×
        let mut refs = Vec::new();
955194×
        for r in o::array(field(m, 0, "refs")?, "refs")? {
956194×
            let rm = map(r, "ref")?;
957194×
            let ls = field(rm, 4, "last serial")?;
958194×
            refs.push(RemoteRef {
959194×
                name: o::text(field(rm, 0, "name")?, "name")?.to_owned(),
960194×
                state: o::oref(field(rm, 1, "state")?, "state")?,
961194×
                heads: hashes_from(field(rm, 2, "heads")?)?,
962194×
                attestation: opt_attest_from(field(rm, 3, "attestation")?)?,
963194×
                last_serial: if ls.is_null() {
964178×
                    None
965
                } else {
96616×
                    Some(o::uint(ls, "last serial")?)
967
                },
968
            });
969
        }
970190×
        let policy = match m.get(&2) {
971190×
            Some(p) => opt_policy_from(p)?,
9720×
            None => None,
973
        };
974190×
        let tags = match m.get(&3) {
975190×
            Some(t) => tags_from(t)?,
9760×
            None => Vec::new(),
977
        };
978190×
        let mut history = Vec::new();
979190×
        if let Some(h) = m.get(&4) {
980190×
            for p in o::array(h, "history")? {
9811×
                history.push(policy_from(p)?);
982
            }
9830×
        }
984190×
        let default_bookmark = match m.get(&5) {
985121×
            Some(Value::Text(t)) => Some(t.clone()),
98669×
            _ => None,
987
        };
988190×
        Ok((
989190×
            id,
990190×
            RefsResponse {
991190×
                refs,
992190×
                policy,
993190×
                tags,
994190×
                history,
995190×
                default_bookmark,
996190×
            },
997190×
        ))
998190×
    }
999
100048×
    pub fn encode_negotiate(have: &[ChangeHash], want: &[ChangeHash]) -> Vec<u8> {
100148×
        cbor::encode(&Value::map([
100248×
            (0, hashes_value(have)),
100348×
            (1, hashes_value(want)),
100448×
        ]))
100548×
    }
100648×
    pub fn decode_negotiate(bytes: &[u8]) -> Result<(Vec<ChangeHash>, Vec<ChangeHash>)> {
100748×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
100848×
        let m = map(&v, "negotiate")?;
1009
        Ok((
101048×
            hashes_from(field(m, 0, "have")?)?,
101148×
            hashes_from(field(m, 1, "want")?)?,
1012
        ))
101348×
    }
1014
1015
    /// `fetch` request: `{0: [changes], 1: max blob size or null}`.
101649×
    pub fn encode_fetch(hs: &[ChangeHash], max_blob: Option<u64>) -> Vec<u8> {
101749×
        cbor::encode(&Value::map([
101849×
            (0, hashes_value(hs)),
101949×
            (1, max_blob.map(Value::Uint).unwrap_or(Value::Null)),
102049×
        ]))
102149×
    }
102250×
    pub fn decode_fetch(bytes: &[u8]) -> Result<(Vec<ChangeHash>, Option<u64>)> {
102350×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
102450×
        let m = map(&v, "fetch")?;
102550×
        let max = match m.get(&1) {
102649×
            Some(x) if !x.is_null() => Some(o::uint(x, "max blob")?),
102748×
            _ => None,
1028
        };
102950×
        Ok((hashes_from(field(m, 0, "changes")?)?, max))
103050×
    }
1031
1032
    /// `shallow` request: `{0: [heads], 1: depth}`.
10331×
    pub fn encode_shallow(hs: &[ChangeHash], depth: u32) -> Vec<u8> {
10341×
        cbor::encode(&Value::map([
10351×
            (0, hashes_value(hs)),
10361×
            (1, Value::Uint(depth as u64)),
10371×
        ]))
10381×
    }
10391×
    pub fn decode_shallow(bytes: &[u8]) -> Result<(Vec<ChangeHash>, u32)> {
10401×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
10411×
        let m = map(&v, "shallow")?;
1042
        Ok((
10431×
            hashes_from(field(m, 0, "heads")?)?,
10441×
            o::u32_(field(m, 1, "depth")?, "depth")?,
1045
        ))
10461×
    }
1047
104850×
    pub fn encode_hashes(hs: &[ChangeHash]) -> Vec<u8> {
104950×
        cbor::encode(&Value::map([(0, hashes_value(hs))]))
105050×
    }
105149×
    pub fn decode_hashes(bytes: &[u8]) -> Result<Vec<ChangeHash>> {
105249×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
105349×
        hashes_from(field(map(&v, "hashes")?, 0, "hashes")?)
105449×
    }
1055
105633×
    pub fn encode_push(
105733×
        bundle: &Bundle,
105833×
        updates: &[RefUpdate],
105933×
        policy: Option<&SignedPolicy>,
106033×
        tags: &[TagRef],
106133×
    ) -> Vec<u8> {
106233×
        cbor::encode(&Value::map([
106333×
            (0, bundle_value(bundle)),
106433×
            (2, opt_policy_value(policy)),
106533×
            (3, tags_value(tags)),
1066
            (
1067
                1,
1068
                Value::Array(
106933×
                    updates
107033×
                        .iter()
107133×
                        .map(|u| {
107233×
                            Value::map([
107333×
                                (0, Value::Text(u.name.clone())),
107433×
                                (1, Value::Bytes(u.old_state.0.to_vec())),
107533×
                                (2, Value::Bytes(u.new_state.0.to_vec())),
107633×
                                (3, Value::Bool(u.force)),
107733×
                                (4, opt_attest_value(&u.attestation)),
107833×
                            ])
107933×
                        })
108033×
                        .collect(),
1081
                ),
1082
            ),
1083
        ]))
108433×
    }
1085
    /// A decoded push: bundle, bookmark updates, policy, tags.
1086
    pub type PushParts = (Bundle, Vec<RefUpdate>, Option<SignedPolicy>, Vec<TagRef>);
1087
108829×
    pub fn decode_push(bytes: &[u8]) -> Result<PushParts> {
108929×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
109029×
        let m = map(&v, "push")?;
109129×
        let bundle = bundle_from_value(field(m, 0, "bundle")?)?;
109229×
        let mut updates = Vec::new();
109329×
        for u in o::array(field(m, 1, "updates")?, "updates")? {
109429×
            let um = map(u, "update")?;
109529×
            updates.push(RefUpdate {
109629×
                name: o::text(field(um, 0, "name")?, "name")?.to_owned(),
109729×
                old_state: o::oref(field(um, 1, "old_state")?, "old_state")?,
109829×
                new_state: o::oref(field(um, 2, "new_state")?, "new_state")?,
109929×
                force: field(um, 3, "force")?
110029×
                    .as_bool()
110129×
                    .ok_or_else(|| schema("force: expected bool"))?,
110229×
                attestation: opt_attest_from(field(um, 4, "attestation")?)?,
1103
            });
1104
        }
110529×
        let policy = match m.get(&2) {
110629×
            Some(p) => opt_policy_from(p)?,
11070×
            None => None,
1108
        };
110929×
        let tags = match m.get(&3) {
111029×
            Some(t) => tags_from(t)?,
11110×
            None => Vec::new(),
1112
        };
111329×
        Ok((bundle, updates, policy, tags))
111429×
    }
1115
111629×
    pub fn encode_push_response(results: &[RefResult]) -> Vec<u8> {
111729×
        cbor::encode(&Value::map([(
1118
            0,
1119
            Value::Array(
112029×
                results
112129×
                    .iter()
112230×
                    .map(|r| {
112330×
                        Value::map([
112430×
                            (0, Value::Text(r.name.clone())),
112530×
                            (1, Value::Bool(r.ok)),
112630×
                            (2, Value::Text(r.reason.clone())),
112730×
                        ])
112830×
                    })
112929×
                    .collect(),
1130
            ),
1131
        )]))
113229×
    }
113329×
    pub fn decode_push_response(bytes: &[u8]) -> Result<Vec<RefResult>> {
113429×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
113529×
        let m = map(&v, "push response")?;
113629×
        let mut out = Vec::new();
113730×
        for r in o::array(field(m, 0, "results")?, "results")? {
113830×
            let rm = map(r, "result")?;
113930×
            out.push(RefResult {
114030×
                name: o::text(field(rm, 0, "name")?, "name")?.to_owned(),
114130×
                ok: field(rm, 1, "ok")?
114230×
                    .as_bool()
114330×
                    .ok_or_else(|| schema("ok: expected bool"))?,
114430×
                reason: o::text(field(rm, 2, "reason")?, "reason")?.to_owned(),
1145
            });
1146
        }
114729×
        Ok(out)
114829×
    }
1149
}
1150
1151
#[cfg(test)]
1152
mod tests {
1153
    use super::*;
1154
    use crate::attest::AttestBody;
1155
    use crate::objects::Hlc;
1156
1157
    #[test]
11581×
    fn wire_round_trips() {
11591×
        let c = crate::objects::tests::sample_change();
11601×
        let sk = crate::identity::SecretKey {
11611×
            ed_seed: [1; 32],
11621×
            ml_seed: [2; 32],
11631×
        };
11641×
        let id = sk.identity("a", "b", 0);
11651×
        let key = id.hash();
11661×
        let sig = sk.sign(crate::identity::Domain::Change, &key, &c.hash());
11671×
        let seal = Attest::seal([3; 16], vec![c.hash()]);
11681×
        let seal_sig = sk.sign(crate::identity::Domain::Attest, &key, &seal.hash());
11691×
        let policy = Policy::single_root([3; 16], id.clone(), 99);
11701×
        let psig = sk.sign(crate::identity::Domain::Policy, &key, &policy.hash());
11711×
        let bundle = Bundle {
11721×
            changes: vec![(c.clone(), vec![sig.clone()])],
11731×
            attests: vec![SignedAttest {
11741×
                attest: seal.clone(),
11751×
                sigs: vec![seal_sig],
11761×
            }],
11771×
            objects: vec![
11781×
                (ObjectType::Identity, id.encode()),
11791×
                (ObjectType::Chunk, b"xyz".to_vec()),
11801×
            ],
11811×
            policies: vec![SignedPolicy {
11821×
                policy: policy.clone(),
11831×
                sigs: vec![psig.clone()],
11841×
            }],
11851×
            hashes: vec![],
11861×
            raw_changes: vec![],
11871×
            certificates: vec![],
11881×
        };
11891×
        let d = wire::decode_bundle(&wire::encode_bundle(&bundle)).unwrap();
11901×
        assert_eq!(
1191
            d.hashes,
11921×
            vec![c.hash()],
1193
            "the receiver hashes what it was sent"
1194
        );
11951×
        assert_eq!(d.changes, bundle.changes);
11961×
        assert_eq!(d.attests, bundle.attests);
11971×
        assert_eq!(d.objects, bundle.objects);
11981×
        assert_eq!(d.policies, bundle.policies);
11991×
        assert!(d.certificates.is_empty());
1200
        // Certificates (keyless, oxygenhub#241) travel as field 4, only
1201
        // when there are any: a bundle without them encodes as before.
12021×
        let plain = wire::encode_bundle(&bundle);
12031×
        let mut with = bundle.clone();
12041×
        let issuer_sk = crate::identity::SecretKey::generate();
12051×
        let issuer = issuer_sk.identity("hub", "h@x", 1);
12061×
        with.certificates.push(crate::cert::SignedCertificate::sign(
12071×
            crate::cert::Certificate {
12081×
                key: id.clone(),
12091×
                issuer: issuer.hash(),
12101×
                subject: "agent:a@b".into(),
12111×
                claims: vec![],
12121×
                scope: crate::identity::Scope::CHANGES,
12131×
                not_before: 1,
12141×
                expires: 2,
12151×
            },
12161×
            &issuer,
12171×
            &issuer_sk,
1218
        ));
12191×
        let encoded = wire::encode_bundle(&with);
12201×
        assert_ne!(encoded, plain);
12211×
        assert_eq!(
12221×
            wire::decode_bundle(&encoded).unwrap().certificates,
1223
            with.certificates
1224
        );
12251×
        with.certificates.clear();
12261×
        assert_eq!(wire::encode_bundle(&with), plain);
1227
12281×
        let att = Attest {
12291×
            repo: [3; 16],
12301×
            hlc: Hlc::default(),
12311×
            body: AttestBody::Ref {
12321×
                name: "main".into(),
12331×
                old_state: ObjectRef([1; 32]),
12341×
                new_state: ObjectRef([2; 32]),
12351×
                serial: 4,
12361×
            },
12371×
        };
12381×
        let refs = vec![
12391×
            RemoteRef {
12401×
                name: "main".into(),
12411×
                state: ObjectRef([2; 32]),
12421×
                heads: vec![c.hash()],
12431×
                attestation: Some(SignedAttest {
12441×
                    attest: att.clone(),
12451×
                    sigs: vec![sig.clone()],
12461×
                }),
12471×
                last_serial: Some(4),
12481×
            },
12491×
            RemoteRef {
12501×
                name: "dev".into(),
12511×
                state: ObjectRef([5; 32]),
12521×
                heads: vec![],
12531×
                attestation: None,
12541×
                last_serial: None,
12551×
            },
1256
        ];
12571×
        let tag = TagRef {
12581×
            name: "v1".into(),
12591×
            state: ObjectRef([2; 32]),
12601×
            heads: vec![c.hash()],
12611×
            attestation: None,
12621×
        };
12631×
        let resp = RefsResponse {
12641×
            refs,
12651×
            policy: Some(SignedPolicy {
12661×
                policy: policy.clone(),
12671×
                sigs: vec![psig.clone()],
12681×
            }),
12691×
            tags: vec![tag.clone()],
12701×
            history: vec![SignedPolicy {
12711×
                policy: policy.clone(),
12721×
                sigs: vec![],
12731×
            }],
12741×
            default_bookmark: Some("develop".into()),
12751×
        };
12761×
        let (id_back, back) = wire::decode_refs(&wire::encode_refs([3; 16], &resp)).unwrap();
12771×
        assert_eq!(id_back, [3; 16]);
12781×
        assert_eq!(back.refs.len(), 2);
12791×
        assert_eq!(back.refs[0].attestation.as_ref().unwrap().attest, att);
12801×
        assert_eq!(back.refs[1].last_serial, None);
12811×
        assert_eq!(back.policy.as_ref().unwrap().policy, policy);
12821×
        assert_eq!(back.history, resp.history);
12831×
        assert_eq!(back.tags, vec![tag.clone()]);
1284
12851×
        let (h, w) = wire::decode_negotiate(&wire::encode_negotiate(
12861×
            &[ObjectRef([1; 32])],
12871×
            &[ObjectRef([2; 32])],
12881×
        ))
12891×
        .unwrap();
12901×
        assert_eq!((h, w), (vec![ObjectRef([1; 32])], vec![ObjectRef([2; 32])]));
12911×
        assert_eq!(
12921×
            wire::decode_hashes(&wire::encode_hashes(&[c.hash()])).unwrap(),
12931×
            vec![c.hash()]
1294
        );
12951×
        assert_eq!(
12961×
            wire::decode_fetch(&wire::encode_fetch(&[c.hash()], Some(4096))).unwrap(),
12971×
            (vec![c.hash()], Some(4096))
1298
        );
12991×
        assert_eq!(
13001×
            wire::decode_fetch(&wire::encode_fetch(&[], None)).unwrap(),
13011×
            (vec![], None)
1302
        );
13031×
        assert_eq!(
13041×
            wire::decode_shallow(&wire::encode_shallow(&[c.hash()], 3)).unwrap(),
13051×
            (vec![c.hash()], 3)
1306
        );
1307
        // An old-style fetch request without the filter field still decodes.
13081×
        assert_eq!(
13091×
            wire::decode_fetch(&wire::encode_hashes(&[c.hash()])).unwrap(),
13101×
            (vec![c.hash()], None)
1311
        );
1312
13131×
        let up = RefUpdate {
13141×
            name: "main".into(),
13151×
            old_state: ObjectRef([1; 32]),
13161×
            new_state: ObjectRef([2; 32]),
13171×
            force: true,
13181×
            attestation: Some(SignedAttest {
13191×
                attest: att,
13201×
                sigs: vec![sig],
13211×
            }),
13221×
        };
13231×
        let sp = SignedPolicy {
13241×
            policy,
13251×
            sigs: vec![psig],
13261×
        };
13271×
        let (b2, ups, pol, tags) = wire::decode_push(&wire::encode_push(
13281×
            &bundle,
13291×
            std::slice::from_ref(&up),
13301×
            Some(&sp),
13311×
            std::slice::from_ref(&tag),
13321×
        ))
13331×
        .unwrap();
13341×
        assert_eq!(pol, Some(sp));
13351×
        assert_eq!(tags, vec![tag]);
13361×
        assert_eq!(b2.changes.len(), 1);
13371×
        assert_eq!(ups[0].name, "main");
13381×
        assert!(ups[0].force);
13391×
        assert_eq!(ups[0].attestation, up.attestation);
13401×
        let res = vec![RefResult {
13411×
            name: "main".into(),
13421×
            ok: false,
13431×
            reason: "stale".into(),
13441×
        }];
13451×
        assert_eq!(
13461×
            wire::decode_push_response(&wire::encode_push_response(&res)).unwrap(),
1347
            res
1348
        );
13491×
        assert_eq!(
13501×
            wire::decode_capabilities(&wire::encode_capabilities([7; 16])).unwrap(),
1351
            [7; 16]
1352
        );
13531×
    }
1354
}