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