OXYGENHUB
# 5259a3af8ee6 oxygen / crates / oxy-core / src / protocol.rs
24.5 KiB · Rust · last changed Lazy blobs: clone --max-blob / filter.max-blob, absent-blob registry fetched on first read, objects endpoint (spec 0.6e) 1 month ago source blame history raw
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;
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
/// What `refs` returns: every bookmark plus the current policy, if any.
37
#[derive(Clone, Debug, Default)]
38
pub struct RefsResponse {
39
    pub refs: Vec<RemoteRef>,
40
    pub policy: Option<SignedPolicy>,
41
}
42
43
/// One reference as a remote reports it.
44
#[derive(Clone, Debug)]
45
pub struct RemoteRef {
46
    pub name: String,
47
    pub state: ObjectRef,
48
    pub heads: Vec<ChangeHash>,
49
    pub attestation: Option<SignedAttest>,
50
}
51
52
/// Everything needed to add a set of changes to a repository.
53
#[derive(Clone, Debug, Default)]
54
pub struct Bundle {
55
    /// In dependency order, each with its individual signatures.
56
    pub changes: Vec<(Change, Vec<Signature>)>,
57
    /// Seals covering the changes.
58
    pub attests: Vec<SignedAttest>,
59
    /// Chunks, blobs, identities, and state objects referenced by the changes.
60
    pub objects: Vec<(ObjectType, Vec<u8>)>,
61
}
62
63
impl Bundle {
64
    pub fn is_empty(&self) -> bool {
65
        self.changes.is_empty() && self.attests.is_empty() && self.objects.is_empty()
66
    }
67
}
68
69
/// A requested bookmark move (spec §10.2, §16.3).
70
#[derive(Clone, Debug)]
71
pub struct RefUpdate {
72
    pub name: String,
73
    pub old_state: ObjectRef,
74
    pub new_state: ObjectRef,
75
    pub force: bool,
76
    pub attestation: Option<SignedAttest>,
77
}
78
79
#[derive(Clone, Debug, PartialEq, Eq)]
80
pub struct RefResult {
81
    pub name: String,
82
    pub ok: bool,
83
    pub reason: String,
84
}
85
86
pub trait Remote {
87
    fn repo_id(&self) -> Result<[u8; 16]>;
88
    fn refs(&self) -> Result<RefsResponse>;
89
    /// Changes in the closure of `want` but not of `have`, in dependency order.
90
    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>>;
91
    /// The changes with their signatures, seals, identities, and blobs; blobs
92
    /// larger than `max_blob` bytes are left out for lazy fetching (spec §13.5).
93
    fn fetch(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle>;
94
    /// Individual objects by reference (spec §16.1 `objects`); a blob comes
95
    /// with its chunks.
96
    fn objects(&self, refs: &[ObjectRef]) -> Result<Vec<(ObjectType, Vec<u8>)>>;
97
    /// Push changes, bookmark updates, and optionally a newer policy. The
98
    /// policy is applied first; a result named `policy` reports it.
99
    fn push(&self, bundle: Bundle, updates: Vec<RefUpdate>, policy: Option<SignedPolicy>) -> Result<Vec<RefResult>>;
100
}
101
102
// ---------------------------------------------------------------- local
103
104
/// Another repository on this machine, served in-process.
105
pub struct LocalRemote {
106
    pub repo: Repo,
107
}
108
109
impl LocalRemote {
110
    pub fn open(path: &Path) -> Result<LocalRemote> {
111
        Ok(LocalRemote { repo: Repo::open(path)? })
112
    }
113
}
114
115
impl Remote for LocalRemote {
116
    fn repo_id(&self) -> Result<[u8; 16]> {
117
        self.repo.id()
118
    }
119
    fn refs(&self) -> Result<RefsResponse> {
120
        self.repo.remote_refs()
121
    }
122
    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>> {
123
        self.repo.negotiate(have, want)
124
    }
125
    fn fetch(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle> {
126
        self.repo.bundle_for_filtered(changes, max_blob)
127
    }
128
    fn objects(&self, refs: &[ObjectRef]) -> Result<Vec<(ObjectType, Vec<u8>)>> {
129
        self.repo.objects_for(refs)
130
    }
131
    fn push(&self, bundle: Bundle, updates: Vec<RefUpdate>, policy: Option<SignedPolicy>) -> Result<Vec<RefResult>> {
132
        self.repo.receive_push(bundle, updates, policy)
133
    }
134
}
135
136
// ---------------------------------------------------------------- http
137
138
/// How to reach a remote: credentials and trust.
139
#[derive(Clone, Debug, Default)]
140
pub struct RemoteOptions {
141
    /// Bearer token for pushes.
142
    pub token: Option<String>,
143
    /// PEM file with extra root certificates to trust for `https://` remotes
144
    /// (self-hosted servers); the system web roots are always trusted.
145
    pub ca: Option<PathBuf>,
146
}
147
148
/// A remote served by `oxy-server` at `<base>/oxy/v1/...`.
149
pub struct HttpRemote {
150
    base: String,
151
    token: Option<String>,
152
    agent: ureq::Agent,
153
}
154
155
/// A TLS client configuration trusting the web roots plus `ca`, if given.
156
fn tls_client_config(ca: Option<&Path>) -> Result<Arc<rustls::ClientConfig>> {
157
    let mut roots = rustls::RootCertStore::empty();
158
    roots.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
159
    if let Some(path) = ca {
160
        let pem = std::fs::read(path).map_err(|e| RepoError::Msg(format!("reading CA file {}: {e}", path.display())))?;
161
        let certs = rustls_pemfile::certs(&mut pem.as_slice())
162
            .collect::<std::result::Result<Vec<_>, _>>()
163
            .map_err(|e| RepoError::Msg(format!("parsing CA file {}: {e}", path.display())))?;
164
        if certs.is_empty() {
165
            return Err(RepoError::Msg(format!("CA file {} holds no certificates", path.display())));
166
        }
167
        for c in certs {
168
            roots.add(c).map_err(|e| RepoError::Msg(format!("CA file {}: {e}", path.display())))?;
169
        }
170
    }
171
    let cfg = rustls::ClientConfig::builder_with_provider(Arc::new(rustls::crypto::ring::default_provider()))
172
        .with_safe_default_protocol_versions()
173
        .map_err(|e| RepoError::Msg(e.to_string()))?
174
        .with_root_certificates(roots)
175
        .with_no_client_auth();
176
    Ok(Arc::new(cfg))
177
}
178
179
impl HttpRemote {
180
    pub fn new(base: &str, options: &RemoteOptions) -> Result<HttpRemote> {
181
        let agent = ureq::AgentBuilder::new().tls_config(tls_client_config(options.ca.as_deref())?).build();
182
        Ok(HttpRemote { base: base.trim_end_matches('/').to_string(), token: options.token.clone(), agent })
183
    }
184
185
    fn call(&self, post: bool, endpoint: &str, body: Option<&[u8]>) -> Result<Vec<u8>> {
186
        let url = format!("{}/oxy/v1/{endpoint}", self.base);
187
        let mut req = if post { self.agent.post(&url) } else { self.agent.get(&url) };
188
        req = req.set("Accept", CONTENT_TYPE);
189
        if let Some(t) = &self.token {
190
            req = req.set("Authorization", &format!("Bearer {t}"));
191
        }
192
        let resp = match body {
193
            Some(b) => req.set("Content-Type", CONTENT_TYPE).send_bytes(b),
194
            None => req.call(),
195
        };
196
        match resp {
197
            Ok(r) => {
198
                let mut out = Vec::new();
199
                r.into_reader().read_to_end(&mut out).map_err(|e| RepoError::Msg(format!("{url}: {e}")))?;
200
                Ok(out)
201
            }
202
            Err(ureq::Error::Status(code, r)) => {
203
                let text = r.into_string().unwrap_or_default();
204
                Err(RepoError::Msg(format!("{url}: HTTP {code}: {}", text.trim())))
205
            }
206
            Err(e) => Err(RepoError::Msg(format!("{url}: {e}"))),
207
        }
208
    }
209
}
210
211
impl Remote for HttpRemote {
212
    fn repo_id(&self) -> Result<[u8; 16]> {
213
        wire::decode_capabilities(&self.call(false, "capabilities", None)?)
214
    }
215
    fn refs(&self) -> Result<RefsResponse> {
216
        Ok(wire::decode_refs(&self.call(false, "refs", None)?)?.1)
217
    }
218
    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>> {
219
        wire::decode_hashes(&self.call(true, "negotiate", Some(&wire::encode_negotiate(have, want)))?)
220
    }
221
    fn fetch(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle> {
222
        wire::decode_bundle(&self.call(true, "fetch", Some(&wire::encode_fetch(changes, max_blob)))?)
223
    }
224
    fn objects(&self, refs: &[ObjectRef]) -> Result<Vec<(ObjectType, Vec<u8>)>> {
225
        Ok(wire::decode_bundle(&self.call(true, "objects", Some(&wire::encode_hashes(refs)))?)?.objects)
226
    }
227
    fn push(&self, bundle: Bundle, updates: Vec<RefUpdate>, policy: Option<SignedPolicy>) -> Result<Vec<RefResult>> {
228
        wire::decode_push_response(&self.call(true, "push", Some(&wire::encode_push(&bundle, &updates, policy.as_ref())))?)
229
    }
230
}
231
232
/// Open a remote by URL: a local path (optionally `file://`), or `http(s)://`.
233
pub fn open_remote(url: &str, options: &RemoteOptions) -> Result<Box<dyn Remote>> {
234
    if url.starts_with("http://") || url.starts_with("https://") {
235
        return Ok(Box::new(HttpRemote::new(url, options)?));
236
    }
237
    let path = url.strip_prefix("file://").unwrap_or(url);
238
    if path.contains("://") {
239
        return Err(RepoError::Msg(format!("unsupported remote URL {url:?}")));
240
    }
241
    Ok(Box::new(LocalRemote::open(Path::new(path))?))
242
}
243
244
// ---------------------------------------------------------------- wire
245
246
/// CBOR encodings of the protocol messages (spec §16.1).
247
pub mod wire {
248
    use super::*;
249
250
    fn schema(msg: &str) -> RepoError {
251
        RepoError::Object(ObjectError::Schema(msg.into()))
252
    }
253
    fn sigs_value(sigs: &[Signature]) -> Value {
254
        Value::Array(sigs.iter().map(|s| s.to_value()).collect())
255
    }
256
    fn sigs_from(v: &Value) -> Result<Vec<Signature>> {
257
        Ok(o::array(v, "signatures")?.iter().map(Signature::from_value).collect::<std::result::Result<Vec<_>, _>>()?)
258
    }
259
    fn hashes_value(hs: &[ObjectRef]) -> Value {
260
        Value::Array(hs.iter().map(|h| Value::Bytes(h.0.to_vec())).collect())
261
    }
262
    fn hashes_from(v: &Value) -> Result<Vec<ObjectRef>> {
263
        Ok(o::array(v, "hashes")?.iter().map(|h| o::oref(h, "hash")).collect::<std::result::Result<Vec<_>, _>>()?)
264
    }
265
    fn attest_value(sa: &SignedAttest) -> Value {
266
        Value::Array(vec![Value::Bytes(sa.attest.encode()), sigs_value(&sa.sigs)])
267
    }
268
    fn attest_from(v: &Value) -> Result<SignedAttest> {
269
        let a = o::array(v, "signed attest")?;
270
        if a.len() != 2 {
271
            return Err(schema("signed attest: expected [attest, sigs]"));
272
        }
273
        Ok(SignedAttest { attest: Attest::decode(o::bytes(&a[0], "attest")?)?, sigs: sigs_from(&a[1])? })
274
    }
275
    fn opt_attest_value(sa: &Option<SignedAttest>) -> Value {
276
        sa.as_ref().map(attest_value).unwrap_or(Value::Null)
277
    }
278
    fn opt_attest_from(v: &Value) -> Result<Option<SignedAttest>> {
279
        if v.is_null() { Ok(None) } else { attest_from(v).map(Some) }
280
    }
281
    fn opt_policy_value(sp: Option<&SignedPolicy>) -> Value {
282
        match sp {
283
            Some(sp) => Value::Array(vec![Value::Bytes(sp.policy.encode()), sigs_value(&sp.sigs)]),
284
            None => Value::Null,
285
        }
286
    }
287
    fn opt_policy_from(v: &Value) -> Result<Option<SignedPolicy>> {
288
        if v.is_null() {
289
            return Ok(None);
290
        }
291
        let a = o::array(v, "signed policy")?;
292
        if a.len() != 2 {
293
            return Err(schema("signed policy: expected [policy, sigs]"));
294
        }
295
        Ok(Some(SignedPolicy { policy: Policy::decode(o::bytes(&a[0], "policy")?)?, sigs: sigs_from(&a[1])? }))
296
    }
297
    fn map<'a>(v: &'a Value, what: &str) -> Result<&'a o::Map> {
298
        Ok(o::map(v, what)?)
299
    }
300
    fn field<'a>(m: &'a o::Map, k: u64, what: &str) -> Result<&'a Value> {
301
        Ok(o::field(m, k, what)?)
302
    }
303
304
    pub fn bundle_value(b: &Bundle) -> Value {
305
        Value::map([
306
            (
307
                0,
308
                Value::Array(b.changes.iter().map(|(c, s)| Value::Array(vec![Value::Bytes(c.encode()), sigs_value(s)])).collect()),
309
            ),
310
            (1, Value::Array(b.attests.iter().map(attest_value).collect())),
311
            (
312
                2,
313
                Value::Array(
314
                    b.objects.iter().map(|(t, body)| Value::Array(vec![Value::Text(t.name().into()), Value::Bytes(body.clone())])).collect(),
315
                ),
316
            ),
317
        ])
318
    }
319
320
    pub fn bundle_from_value(v: &Value) -> Result<Bundle> {
321
        let m = map(v, "bundle")?;
322
        let mut b = Bundle::default();
323
        for c in o::array(field(m, 0, "changes")?, "changes")? {
324
            let pair = o::array(c, "change entry")?;
325
            if pair.len() != 2 {
326
                return Err(schema("change entry: expected [change, sigs]"));
327
            }
328
            b.changes.push((Change::decode(o::bytes(&pair[0], "change")?)?, sigs_from(&pair[1])?));
329
        }
330
        for a in o::array(field(m, 1, "attests")?, "attests")? {
331
            b.attests.push(attest_from(a)?);
332
        }
333
        for ob in o::array(field(m, 2, "objects")?, "objects")? {
334
            let pair = o::array(ob, "object")?;
335
            if pair.len() != 2 {
336
                return Err(schema("object: expected [type, body]"));
337
            }
338
            let ty = ObjectType::from_name(o::text(&pair[0], "object type")?).ok_or_else(|| schema("unknown object type"))?;
339
            b.objects.push((ty, o::bytes(&pair[1], "object body")?.to_vec()));
340
        }
341
        Ok(b)
342
    }
343
344
    pub fn encode_bundle(b: &Bundle) -> Vec<u8> {
345
        cbor::encode(&bundle_value(b))
346
    }
347
    pub fn decode_bundle(bytes: &[u8]) -> Result<Bundle> {
348
        bundle_from_value(&cbor::decode(bytes).map_err(ObjectError::from)?)
349
    }
350
351
    pub fn encode_capabilities(id: [u8; 16]) -> Vec<u8> {
352
        cbor::encode(&Value::map([
353
            (0, Value::Array(vec![Value::Uint(PROTOCOL_VERSION)])),
354
            (1, Value::Bytes(id.to_vec())),
355
            (2, Value::Array(vec![Value::Text("seal".into())])),
356
        ]))
357
    }
358
    pub fn decode_capabilities(bytes: &[u8]) -> Result<[u8; 16]> {
359
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
360
        let m = map(&v, "capabilities")?;
361
        let versions = o::array(field(m, 0, "versions")?, "versions")?;
362
        if !versions.iter().any(|x| x.as_uint() == Some(PROTOCOL_VERSION)) {
363
            return Err(RepoError::Msg("remote does not speak protocol version 1".into()));
364
        }
365
        o::bytes(field(m, 1, "repo id")?, "repo id")?.try_into().map_err(|_| schema("repo id: expected 16 bytes"))
366
    }
367
368
    pub fn encode_refs(id: [u8; 16], resp: &RefsResponse) -> Vec<u8> {
369
        cbor::encode(&Value::map([
370
            (
371
                0,
372
                Value::Array(
373
                    resp.refs
374
                        .iter()
375
                        .map(|r| {
376
                            Value::map([
377
                                (0, Value::Text(r.name.clone())),
378
                                (1, Value::Bytes(r.state.0.to_vec())),
379
                                (2, hashes_value(&r.heads)),
380
                                (3, opt_attest_value(&r.attestation)),
381
                                (4, r.last_serial.map(Value::Uint).unwrap_or(Value::Null)),
382
                            ])
383
                        })
384
                        .collect(),
385
                ),
386
            ),
387
            (1, Value::Bytes(id.to_vec())),
388
            (2, opt_policy_value(resp.policy.as_ref())),
389
        ]))
390
    }
391
    pub fn decode_refs(bytes: &[u8]) -> Result<([u8; 16], RefsResponse)> {
392
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
393
        let m = map(&v, "refs")?;
394
        let id: [u8; 16] = o::bytes(field(m, 1, "repo id")?, "repo id")?.try_into().map_err(|_| schema("repo id: expected 16 bytes"))?;
395
        let mut refs = Vec::new();
396
        for r in o::array(field(m, 0, "refs")?, "refs")? {
397
            let rm = map(r, "ref")?;
398
            let ls = field(rm, 4, "last serial")?;
399
            refs.push(RemoteRef {
400
                name: o::text(field(rm, 0, "name")?, "name")?.to_owned(),
401
                state: o::oref(field(rm, 1, "state")?, "state")?,
402
                heads: hashes_from(field(rm, 2, "heads")?)?,
403
                attestation: opt_attest_from(field(rm, 3, "attestation")?)?,
404
                last_serial: if ls.is_null() { None } else { Some(o::uint(ls, "last serial")?) },
405
            });
406
        }
407
        let policy = match m.get(&2) {
408
            Some(p) => opt_policy_from(p)?,
409
            None => None,
410
        };
411
        Ok((id, RefsResponse { refs, policy }))
412
    }
413
414
    pub fn encode_negotiate(have: &[ChangeHash], want: &[ChangeHash]) -> Vec<u8> {
415
        cbor::encode(&Value::map([(0, hashes_value(have)), (1, hashes_value(want))]))
416
    }
417
    pub fn decode_negotiate(bytes: &[u8]) -> Result<(Vec<ChangeHash>, Vec<ChangeHash>)> {
418
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
419
        let m = map(&v, "negotiate")?;
420
        Ok((hashes_from(field(m, 0, "have")?)?, hashes_from(field(m, 1, "want")?)?))
421
    }
422
423
    /// `fetch` request: `{0: [changes], 1: max blob size or null}`.
424
    pub fn encode_fetch(hs: &[ChangeHash], max_blob: Option<u64>) -> Vec<u8> {
425
        cbor::encode(&Value::map([(0, hashes_value(hs)), (1, max_blob.map(Value::Uint).unwrap_or(Value::Null))]))
426
    }
427
    pub fn decode_fetch(bytes: &[u8]) -> Result<(Vec<ChangeHash>, Option<u64>)> {
428
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
429
        let m = map(&v, "fetch")?;
430
        let max = match m.get(&1) {
431
            Some(x) if !x.is_null() => Some(o::uint(x, "max blob")?),
432
            _ => None,
433
        };
434
        Ok((hashes_from(field(m, 0, "changes")?)?, max))
435
    }
436
437
    pub fn encode_hashes(hs: &[ChangeHash]) -> Vec<u8> {
438
        cbor::encode(&Value::map([(0, hashes_value(hs))]))
439
    }
440
    pub fn decode_hashes(bytes: &[u8]) -> Result<Vec<ChangeHash>> {
441
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
442
        hashes_from(field(map(&v, "hashes")?, 0, "hashes")?)
443
    }
444
445
    pub fn encode_push(bundle: &Bundle, updates: &[RefUpdate], policy: Option<&SignedPolicy>) -> Vec<u8> {
446
        cbor::encode(&Value::map([
447
            (0, bundle_value(bundle)),
448
            (2, opt_policy_value(policy)),
449
            (
450
                1,
451
                Value::Array(
452
                    updates
453
                        .iter()
454
                        .map(|u| {
455
                            Value::map([
456
                                (0, Value::Text(u.name.clone())),
457
                                (1, Value::Bytes(u.old_state.0.to_vec())),
458
                                (2, Value::Bytes(u.new_state.0.to_vec())),
459
                                (3, Value::Bool(u.force)),
460
                                (4, opt_attest_value(&u.attestation)),
461
                            ])
462
                        })
463
                        .collect(),
464
                ),
465
            ),
466
        ]))
467
    }
468
    pub fn decode_push(bytes: &[u8]) -> Result<(Bundle, Vec<RefUpdate>, Option<SignedPolicy>)> {
469
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
470
        let m = map(&v, "push")?;
471
        let bundle = bundle_from_value(field(m, 0, "bundle")?)?;
472
        let mut updates = Vec::new();
473
        for u in o::array(field(m, 1, "updates")?, "updates")? {
474
            let um = map(u, "update")?;
475
            updates.push(RefUpdate {
476
                name: o::text(field(um, 0, "name")?, "name")?.to_owned(),
477
                old_state: o::oref(field(um, 1, "old_state")?, "old_state")?,
478
                new_state: o::oref(field(um, 2, "new_state")?, "new_state")?,
479
                force: field(um, 3, "force")?.as_bool().ok_or_else(|| schema("force: expected bool"))?,
480
                attestation: opt_attest_from(field(um, 4, "attestation")?)?,
481
            });
482
        }
483
        let policy = match m.get(&2) {
484
            Some(p) => opt_policy_from(p)?,
485
            None => None,
486
        };
487
        Ok((bundle, updates, policy))
488
    }
489
490
    pub fn encode_push_response(results: &[RefResult]) -> Vec<u8> {
491
        cbor::encode(&Value::map([(
492
            0,
493
            Value::Array(
494
                results
495
                    .iter()
496
                    .map(|r| Value::map([(0, Value::Text(r.name.clone())), (1, Value::Bool(r.ok)), (2, Value::Text(r.reason.clone()))]))
497
                    .collect(),
498
            ),
499
        )]))
500
    }
501
    pub fn decode_push_response(bytes: &[u8]) -> Result<Vec<RefResult>> {
502
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
503
        let m = map(&v, "push response")?;
504
        let mut out = Vec::new();
505
        for r in o::array(field(m, 0, "results")?, "results")? {
506
            let rm = map(r, "result")?;
507
            out.push(RefResult {
508
                name: o::text(field(rm, 0, "name")?, "name")?.to_owned(),
509
                ok: field(rm, 1, "ok")?.as_bool().ok_or_else(|| schema("ok: expected bool"))?,
510
                reason: o::text(field(rm, 2, "reason")?, "reason")?.to_owned(),
511
            });
512
        }
513
        Ok(out)
514
    }
515
}
516
517
#[cfg(test)]
518
mod tests {
519
    use super::*;
520
    use crate::attest::AttestBody;
521
    use crate::objects::Hlc;
522
523
    #[test]
524
    fn wire_round_trips() {
525
        let c = crate::objects::tests::sample_change();
526
        let sk = crate::identity::SecretKey { ed_seed: [1; 32], ml_seed: [2; 32] };
527
        let id = sk.identity("a", "b", 0);
528
        let key = id.hash();
529
        let sig = sk.sign(crate::identity::Domain::Change, &key, &c.hash());
530
        let seal = Attest::seal([3; 16], vec![c.hash()]);
531
        let seal_sig = sk.sign(crate::identity::Domain::Attest, &key, &seal.hash());
532
        let bundle = Bundle {
533
            changes: vec![(c.clone(), vec![sig.clone()])],
534
            attests: vec![SignedAttest { attest: seal.clone(), sigs: vec![seal_sig] }],
535
            objects: vec![(ObjectType::Identity, id.encode()), (ObjectType::Chunk, b"xyz".to_vec())],
536
        };
537
        let d = wire::decode_bundle(&wire::encode_bundle(&bundle)).unwrap();
538
        assert_eq!(d.changes, bundle.changes);
539
        assert_eq!(d.attests, bundle.attests);
540
        assert_eq!(d.objects, bundle.objects);
541
542
        let att = Attest { repo: [3; 16], hlc: Hlc::default(), body: AttestBody::Ref { name: "main".into(), old_state: ObjectRef([1; 32]), new_state: ObjectRef([2; 32]), serial: 4 } };
543
        let refs = vec![
544
            RemoteRef { name: "main".into(), state: ObjectRef([2; 32]), heads: vec![c.hash()], attestation: Some(SignedAttest { attest: att.clone(), sigs: vec![sig.clone()] }), last_serial: Some(4) },
545
            RemoteRef { name: "dev".into(), state: ObjectRef([5; 32]), heads: vec![], attestation: None, last_serial: None },
546
        ];
547
        let policy = Policy::single_root([3; 16], id.clone(), 99);
548
        let psig = sk.sign(crate::identity::Domain::Policy, &key, &policy.hash());
549
        let resp = RefsResponse { refs, policy: Some(SignedPolicy { policy: policy.clone(), sigs: vec![psig.clone()] }) };
550
        let (id_back, back) = wire::decode_refs(&wire::encode_refs([3; 16], &resp)).unwrap();
551
        assert_eq!(id_back, [3; 16]);
552
        assert_eq!(back.refs.len(), 2);
553
        assert_eq!(back.refs[0].attestation.as_ref().unwrap().attest, att);
554
        assert_eq!(back.refs[1].last_serial, None);
555
        assert_eq!(back.policy.as_ref().unwrap().policy, policy);
556
557
        let (h, w) = wire::decode_negotiate(&wire::encode_negotiate(&[ObjectRef([1; 32])], &[ObjectRef([2; 32])])).unwrap();
558
        assert_eq!((h, w), (vec![ObjectRef([1; 32])], vec![ObjectRef([2; 32])]));
559
        assert_eq!(wire::decode_hashes(&wire::encode_hashes(&[c.hash()])).unwrap(), vec![c.hash()]);
560
        assert_eq!(wire::decode_fetch(&wire::encode_fetch(&[c.hash()], Some(4096))).unwrap(), (vec![c.hash()], Some(4096)));
561
        assert_eq!(wire::decode_fetch(&wire::encode_fetch(&[], None)).unwrap(), (vec![], None));
562
        // An old-style fetch request without the filter field still decodes.
563
        assert_eq!(wire::decode_fetch(&wire::encode_hashes(&[c.hash()])).unwrap(), (vec![c.hash()], None));
564
565
        let up = RefUpdate { name: "main".into(), old_state: ObjectRef([1; 32]), new_state: ObjectRef([2; 32]), force: true, attestation: Some(SignedAttest { attest: att, sigs: vec![sig] }) };
566
        let sp = SignedPolicy { policy, sigs: vec![psig] };
567
        let (b2, ups, pol) = wire::decode_push(&wire::encode_push(&bundle, std::slice::from_ref(&up), Some(&sp))).unwrap();
568
        assert_eq!(pol, Some(sp));
569
        assert_eq!(b2.changes.len(), 1);
570
        assert_eq!(ups[0].name, "main");
571
        assert!(ups[0].force);
572
        assert_eq!(ups[0].attestation, up.attestation);
573
        let res = vec![RefResult { name: "main".into(), ok: false, reason: "stale".into() }];
574
        assert_eq!(wire::decode_push_response(&wire::encode_push_response(&res)).unwrap(), res);
575
        assert_eq!(wire::decode_capabilities(&wire::encode_capabilities([7; 16])).unwrap(), [7; 16]);
576
    }
577
}