OXYGENHUB
# ef7d4ea6f33b oxygen / crates / oxy-core / src / protocol.rs
50.7 KiB · Rust · last changed Grants let certified keys act for an author (keyless step 5a) 18 days 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, 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`).
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(
43
            &crate::cbor::decode(b).map_err(|e| RepoError::Object(ObjectError::from(e)))?,
44
        )
45
    }
46
}
47
48
/// What `refs` returns: every bookmark, every tag, and the current policy.
49
#[derive(Clone, Debug, Default)]
50
pub struct RefsResponse {
51
    pub refs: Vec<RemoteRef>,
52
    pub policy: Option<SignedPolicy>,
53
    pub tags: Vec<TagRef>,
54
    /// Earlier policies, ascending by serial, so a client behind by more
55
    /// than one update can verify each step (spec §14.5).
56
    pub history: Vec<SignedPolicy>,
57
    /// The bookmark a clone checks out (spec §16.1), when the remote names one.
58
    pub default_bookmark: Option<String>,
59
}
60
61
/// A tag as a remote reports it (spec §10.1).
62
#[derive(Clone, Debug, PartialEq, Eq)]
63
pub struct TagRef {
64
    pub name: String,
65
    pub state: ObjectRef,
66
    pub heads: Vec<ChangeHash>,
67
    pub attestation: Option<SignedAttest>,
68
}
69
70
/// One reference as a remote reports it.
71
#[derive(Clone, Debug)]
72
pub struct RemoteRef {
73
    pub name: String,
74
    pub state: ObjectRef,
75
    pub heads: Vec<ChangeHash>,
76
    /// The latest attestation, only when it describes `state`. An unattested
77
    /// push (accepted while no trust policy exists) leaves it `None`.
78
    pub attestation: Option<SignedAttest>,
79
    /// Serial of the latest attestation ever recorded, stale or not, so the
80
    /// next attestation continues the sequence.
81
    pub last_serial: Option<u64>,
82
    /// The host's acceptance of `attestation` when a short-lived key signed
83
    /// it (spec §14.7): the time its certificate is checked at.
84
    pub accepted: Option<SignedAttest>,
85
}
86
87
/// Everything needed to add a set of changes to a repository.
88
#[derive(Clone, Debug, Default)]
89
pub struct Bundle {
90
    /// In dependency order, each with its individual signatures.
91
    pub changes: Vec<(Change, Vec<Signature>)>,
92
    /// Seals covering the changes.
93
    pub attests: Vec<SignedAttest>,
94
    /// Chunks, blobs, identities, and state objects referenced by the changes.
95
    pub objects: Vec<(ObjectType, Vec<u8>)>,
96
    /// Policies between the receiver's and the one sent with the push,
97
    /// ascending by serial (spec §14.5).
98
    pub policies: Vec<SignedPolicy>,
99
    /// Each change's hash, in the order of `changes`, when whoever made
100
    /// the bundle knows it: the store it came from, or the bytes it was
101
    /// decoded from. Never read from the wire. Empty when unknown.
102
    pub hashes: Vec<ChangeHash>,
103
    /// Changes as stored, with their signatures, in dependency order, for a
104
    /// fetch between repositories on one machine: their text blobs and
105
    /// chunks are among `objects`. Never on the wire.
106
    pub raw_changes: Vec<(Vec<u8>, Vec<Signature>)>,
107
    /// Certificates of short-lived keys that signed what the bundle carries
108
    /// (spec §14.7), their issuers' identities among `objects`.
109
    pub certificates: Vec<crate::cert::SignedCertificate>,
110
}
111
112
impl Bundle {
113
    pub fn is_empty(&self) -> bool {
114
        self.changes.is_empty() && self.attests.is_empty() && self.objects.is_empty()
115
    }
116
}
117
118
/// A requested bookmark move (spec §10.2, §16.3).
119
#[derive(Clone, Debug)]
120
pub struct RefUpdate {
121
    pub name: String,
122
    pub old_state: ObjectRef,
123
    pub new_state: ObjectRef,
124
    pub force: bool,
125
    pub attestation: Option<SignedAttest>,
126
}
127
128
#[derive(Clone, Debug, PartialEq, Eq)]
129
pub struct RefResult {
130
    pub name: String,
131
    pub ok: bool,
132
    pub reason: String,
133
}
134
135
pub trait Remote {
136
    fn repo_id(&self) -> Result<[u8; 16]>;
137
    fn refs(&self) -> Result<RefsResponse>;
138
    /// Changes in the closure of `want` but not of `have`, in dependency order.
139
    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>>;
140
    /// The changes with their signatures, seals, identities, and blobs; blobs
141
    /// larger than `max_blob` bytes are left out for lazy fetching (spec §13.5).
142
    fn fetch(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle>;
143
    /// Individual objects by reference (spec §16.1 `objects`); a blob comes
144
    /// with its chunks.
145
    fn objects(&self, refs: &[ObjectRef]) -> Result<Vec<(ObjectType, Vec<u8>)>>;
146
    /// A checkpoint of exactly `heads` (spec §16.1 `checkpoint`): the
147
    /// checkpoint object, its payload blob, and the blob's chunks.
148
    fn checkpoint(&self, heads: &[ChangeHash]) -> Result<Bundle>;
149
    /// The changes within `depth` dependency steps of `heads` (spec §17).
150
    fn shallow(&self, heads: &[ChangeHash], depth: u32) -> Result<Vec<ChangeHash>>;
151
    /// Push changes, bookmark updates, new tags, and optionally a newer
152
    /// policy. The policy is applied first; a result named `policy` reports
153
    /// it, and each tag reports as `tags/<name>`.
154
    fn push(
155
        &self,
156
        bundle: Bundle,
157
        updates: Vec<RefUpdate>,
158
        policy: Option<SignedPolicy>,
159
        tags: Vec<TagRef>,
160
    ) -> Result<Vec<RefResult>>;
161
}
162
163
// ---------------------------------------------------------------- local
164
165
/// Another repository on this machine, served in-process.
166
pub struct LocalRemote {
167
    pub repo: Repo,
168
    /// The pushing repository's id, which binds an unbound target on the
169
    /// first push, as `Oxy-Repo-Id` does over HTTP.
170
    pub claimed_id: Option<[u8; 16]>,
171
}
172
173
impl LocalRemote {
174
    pub fn open(path: &Path) -> Result<LocalRemote> {
175
        Ok(LocalRemote {
176
            repo: Repo::open(path)?,
177
            claimed_id: None,
178
        })
179
    }
180
}
181
182
impl Remote for LocalRemote {
183
    fn repo_id(&self) -> Result<[u8; 16]> {
184
        self.repo.id()
185
    }
186
    fn refs(&self) -> Result<RefsResponse> {
187
        self.repo.remote_refs()
188
    }
189
    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>> {
190
        self.repo.negotiate(have, want)
191
    }
192
    fn fetch(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle> {
193
        self.repo.bundle_raw_for(changes, max_blob)
194
    }
195
    fn objects(&self, refs: &[ObjectRef]) -> Result<Vec<(ObjectType, Vec<u8>)>> {
196
        self.repo.objects_for(refs)
197
    }
198
    fn checkpoint(&self, heads: &[ChangeHash]) -> Result<Bundle> {
199
        self.repo.checkpoint_bundle_for(heads)
200
    }
201
    fn shallow(&self, heads: &[ChangeHash], depth: u32) -> Result<Vec<ChangeHash>> {
202
        self.repo.shallow_changes(heads, depth)
203
    }
204
    fn push(
205
        &self,
206
        bundle: Bundle,
207
        updates: Vec<RefUpdate>,
208
        policy: Option<SignedPolicy>,
209
        tags: Vec<TagRef>,
210
    ) -> Result<Vec<RefResult>> {
211
        if self.repo.is_unbound() {
212
            match self.claimed_id {
213
                Some(id) => self.repo.bind_id(id)?,
214
                None => {
215
                    return Err(RepoError::Msg(
216
                        "the remote is unbound and this client did not say which repository it is"
217
                            .into(),
218
                    ))
219
                }
220
            }
221
        }
222
        self.repo.receive_push(bundle, updates, policy, tags)
223
    }
224
}
225
226
// ---------------------------------------------------------------- http
227
228
/// How to reach a remote: credentials and trust.
229
#[derive(Clone, Debug, Default)]
230
pub struct RemoteOptions {
231
    /// Bearer token for pushes.
232
    pub token: Option<String>,
233
    /// PEM file with extra root certificates to trust for `https://` remotes
234
    /// (self-hosted servers); the system web roots are always trusted.
235
    pub ca: Option<PathBuf>,
236
    /// The local repository's id, sent as `Oxy-Repo-Id` so an unbound
237
    /// hosted repository can adopt it on the first push (spec §16.1).
238
    pub repo_id: Option<[u8; 16]>,
239
    /// Where a remote kept on a store (`dir://`, `s3://`, …) keeps its
240
    /// mirror and what it has seen: the remote's directory under
241
    /// `.oxy/remotes`. None (a clone, before there is a repository) means
242
    /// a temporary directory.
243
    pub cache_dir: Option<PathBuf>,
244
    /// Other tokens for the same hub, tried in turn when `token` is
245
    /// refused (401, 403 or 404): those of enclosing checkouts, when
246
    /// `token` came from the credential store rather than from this
247
    /// remote's own configuration (oxygen#112).
248
    pub fallback_tokens: Vec<String>,
249
}
250
251
/// Opens a store for a URL it understands, or passes (`None`). Stores that
252
/// need more than the standard library (the cloud ones) are added by the
253
/// program that links them, with [`register_store_opener`].
254
pub type StoreOpener =
255
    fn(&str, &RemoteOptions) -> Option<crate::strata::Result<Box<dyn crate::strata::Store>>>;
256
257
static STORE_OPENERS: Mutex<Vec<StoreOpener>> = Mutex::new(Vec::new());
258
259
pub fn register_store_opener(opener: StoreOpener) {
260
    STORE_OPENERS
261
        .lock()
262
        .unwrap_or_else(|p| p.into_inner())
263
        .push(opener);
264
}
265
266
/// A store for `url`, from `dir://` or any registered opener.
267
pub fn open_store(
268
    url: &str,
269
    options: &RemoteOptions,
270
) -> Option<crate::strata::Result<Box<dyn crate::strata::Store>>> {
271
    if let Some(path) = url.strip_prefix("dir://") {
272
        return Some(
273
            crate::strata::LocalStore::open(path)
274
                .map(|s| Box::new(s) as Box<dyn crate::strata::Store>),
275
        );
276
    }
277
    let openers = STORE_OPENERS.lock().unwrap_or_else(|p| p.into_inner());
278
    openers.iter().find_map(|open| open(url, options))
279
}
280
281
/// The request header carrying the client's repository id.
282
pub const REPO_ID_HEADER: &str = "Oxy-Repo-Id";
283
284
/// The id of a hosted repository that has not adopted one yet.
285
pub const UNBOUND_ID: [u8; 16] = [0; 16];
286
287
/// A remote served by `oxy-server` at `<base>/oxy/v1/...`.
288
pub struct HttpRemote {
289
    base: String,
290
    /// The token in use, and those still to try when it is refused.
291
    token: Mutex<Option<String>>,
292
    fallbacks: Mutex<Vec<String>>,
293
    repo_id: Option<[u8; 16]>,
294
    agent: ureq::Agent,
295
}
296
297
/// A TLS client configuration trusting the web roots plus `ca`, if given.
298
fn tls_client_config(ca: Option<&Path>) -> Result<Arc<rustls::ClientConfig>> {
299
    let mut roots = rustls::RootCertStore::empty();
300
    roots.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
301
    if let Some(path) = ca {
302
        let pem = std::fs::read(path)
303
            .map_err(|e| RepoError::Msg(format!("reading CA file {}: {e}", path.display())))?;
304
        let certs = rustls_pemfile::certs(&mut pem.as_slice())
305
            .collect::<std::result::Result<Vec<_>, _>>()
306
            .map_err(|e| RepoError::Msg(format!("parsing CA file {}: {e}", path.display())))?;
307
        if certs.is_empty() {
308
            return Err(RepoError::Msg(format!(
309
                "CA file {} holds no certificates",
310
                path.display()
311
            )));
312
        }
313
        for c in certs {
314
            roots
315
                .add(c)
316
                .map_err(|e| RepoError::Msg(format!("CA file {}: {e}", path.display())))?;
317
        }
318
    }
319
    let cfg = rustls::ClientConfig::builder_with_provider(Arc::new(
320
        rustls::crypto::ring::default_provider(),
321
    ))
322
    .with_safe_default_protocol_versions()
323
    .map_err(|e| RepoError::Msg(e.to_string()))?
324
    .with_root_certificates(roots)
325
    .with_no_client_auth();
326
    Ok(Arc::new(cfg))
327
}
328
329
impl HttpRemote {
330
    pub fn new(base: &str, options: &RemoteOptions) -> Result<HttpRemote> {
331
        refuse_plain_token(base, options)?;
332
        let agent = ureq::AgentBuilder::new()
333
            .tls_config(tls_client_config(options.ca.as_deref())?)
334
            .build();
335
        Ok(HttpRemote {
336
            base: base.trim_end_matches('/').to_string(),
337
            token: Mutex::new(options.token.clone()),
338
            fallbacks: Mutex::new(options.fallback_tokens.iter().rev().cloned().collect()),
339
            repo_id: options.repo_id,
340
            agent,
341
        })
342
    }
343
344
    /// A request, tried again with the next fallback token while the
345
    /// hub refuses the one in use (oxygen#112).
346
    fn call(&self, post: bool, endpoint: &str, body: Option<&[u8]>) -> Result<Vec<u8>> {
347
        loop {
348
            let token = self.token.lock().unwrap_or_else(|p| p.into_inner()).clone();
349
            match self.call_once(post, endpoint, body, token.as_deref()) {
350
                Err(RepoError::Msg(m)) if token.is_some() && refused(&m) => {
351
                    let next = self
352
                        .fallbacks
353
                        .lock()
354
                        .unwrap_or_else(|p| p.into_inner())
355
                        .pop();
356
                    let Some(next) = next else {
357
                        return Err(RepoError::Msg(m));
358
                    };
359
                    crate::credentials::warn_once(&format!(
360
                        "{}: the stored token was refused; using the one an enclosing checkout has for this hub",
361
                        self.base
362
                    ));
363
                    *self.token.lock().unwrap_or_else(|p| p.into_inner()) = Some(next);
364
                }
365
                other => return other,
366
            }
367
        }
368
    }
369
370
    fn call_once(
371
        &self,
372
        post: bool,
373
        endpoint: &str,
374
        body: Option<&[u8]>,
375
        token: Option<&str>,
376
    ) -> Result<Vec<u8>> {
377
        let url = format!("{}/oxy/v1/{endpoint}", self.base);
378
        let mut req = if post {
379
            self.agent.post(&url)
380
        } else {
381
            self.agent.get(&url)
382
        };
383
        req = req.set("Accept", CONTENT_TYPE);
384
        if let Some(t) = token {
385
            req = req.set("Authorization", &format!("Bearer {t}"));
386
        }
387
        if let Some(id) = &self.repo_id {
388
            req = req.set(REPO_ID_HEADER, &hex::encode(id));
389
        }
390
        let resp = match body {
391
            Some(b) => req.set("Content-Type", CONTENT_TYPE).send_bytes(b),
392
            None => req.call(),
393
        };
394
        match resp {
395
            Ok(r) => {
396
                let mut out = Vec::new();
397
                r.into_reader()
398
                    .read_to_end(&mut out)
399
                    .map_err(|e| RepoError::Msg(format!("{url}: {e}")))?;
400
                Ok(out)
401
            }
402
            Err(ureq::Error::Status(code, r)) => {
403
                let text = r.into_string().unwrap_or_default();
404
                let msg = format!("{url}: HTTP {code}: {}", text.trim());
405
                Err(RepoError::Msg(with_hint(msg)))
406
            }
407
            // The transport error already names the URL.
408
            Err(e) => Err(RepoError::Msg(with_hint(e.to_string()))),
409
        }
410
    }
411
}
412
413
/// Whether a failed request's message says the hub refused its token:
414
/// HTTP 401, 403, or 404 (a hub hides a private repository from a token
415
/// that may not read it).
416
fn refused(message: &str) -> bool {
417
    [": HTTP 401", ": HTTP 403", ": HTTP 404"]
418
        .iter()
419
        .any(|c| message.contains(c))
420
}
421
422
/// Refuse to send a token where anyone on the way could read it: plain
423
/// `http` to another machine (see [`crate::credentials::token_allowed`]).
424
fn refuse_plain_token(url: &str, options: &RemoteOptions) -> Result<()> {
425
    if options.token.is_some() && !crate::credentials::token_allowed(url) {
426
        return Err(RepoError::Msg(format!(
427
            "{url}: refusing to send a token over plain http; use https"
428
        )));
429
    }
430
    Ok(())
431
}
432
433
/// What to do about a failed request, recognised from its message.
434
pub fn hint_for(message: &str) -> Option<&'static str> {
435
    let m = message.to_ascii_lowercase();
436
    let hint = if m.contains("http 401") {
437
        "the hub needs you signed in, or did not accept the token: run `oxy auth login HUB` (pushing needs repo:write)"
438
    } else if m.contains("http 403") {
439
        "the token or account lacks permission: pushing needs a token with repo:write and write access to the repository"
440
    } else if m.contains("http 404") {
441
        "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`)"
442
    } else if m.contains("unknownissuer")
443
        || m.contains("unknown issuer")
444
        || m.contains("certificate")
445
        || m.contains("invalidcertificate")
446
    {
447
        "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`"
448
    } else if m.contains("connection refused") {
449
        "nothing is listening there: check the URL and port, and that the server is running"
450
    } else if m.contains("dns")
451
        || m.contains("failed to lookup")
452
        || m.contains("name or service not known")
453
    {
454
        "the host name does not resolve: check the URL"
455
    } else if m.contains("timed out") || m.contains("timeout") {
456
        "the server did not answer in time: check the network, or try again"
457
    } else {
458
        return None;
459
    };
460
    Some(hint)
461
}
462
463
/// `message`, followed by what to do about it when that is known.
464
pub fn with_hint(message: String) -> String {
465
    match hint_for(&message) {
466
        Some(h) => format!("{message}\n  hint: {h}"),
467
        None => message,
468
    }
469
}
470
471
/// A JSON request to a hub API (`GET`, `POST` or `DELETE`), with the same
472
/// TLS trust and token rules as remotes. Returns the status and body, an
473
/// error status included; only transport failures are errors.
474
pub fn http_json(
475
    method: &str,
476
    url: &str,
477
    body: Option<&str>,
478
    options: &RemoteOptions,
479
) -> Result<(u16, String)> {
480
    refuse_plain_token(url, options)?;
481
    let agent = ureq::AgentBuilder::new()
482
        .tls_config(tls_client_config(options.ca.as_deref())?)
483
        .timeout(std::time::Duration::from_secs(60))
484
        .build();
485
    let mut req = agent.request(method, url).set("Accept", "application/json");
486
    if let Some(t) = &options.token {
487
        req = req.set("Authorization", &format!("Bearer {t}"));
488
    }
489
    let resp = match body {
490
        Some(b) => req.set("Content-Type", "application/json").send_string(b),
491
        None => req.call(),
492
    };
493
    match resp {
494
        Ok(r) => {
495
            let code = r.status();
496
            Ok((code, r.into_string().unwrap_or_default()))
497
        }
498
        Err(ureq::Error::Status(code, r)) => Ok((code, r.into_string().unwrap_or_default())),
499
        Err(e) => Err(RepoError::Msg(with_hint(e.to_string()))),
500
    }
501
}
502
503
/// An authenticated GET of any URL on a host, with the same TLS trust and
504
/// token as remotes use (for hub APIs: proposals). Returns the body.
505
pub fn http_get(url: &str, options: &RemoteOptions) -> Result<Vec<u8>> {
506
    refuse_plain_token(url, options)?;
507
    let agent = ureq::AgentBuilder::new()
508
        .tls_config(tls_client_config(options.ca.as_deref())?)
509
        .build();
510
    let mut req = agent.get(url).set("Accept", "application/json");
511
    if let Some(t) = &options.token {
512
        req = req.set("Authorization", &format!("Bearer {t}"));
513
    }
514
    match req.call() {
515
        Ok(r) => {
516
            let mut out = Vec::new();
517
            r.into_reader()
518
                .read_to_end(&mut out)
519
                .map_err(|e| RepoError::Msg(format!("{url}: {e}")))?;
520
            Ok(out)
521
        }
522
        Err(ureq::Error::Status(code, r)) => {
523
            let text = r.into_string().unwrap_or_default();
524
            Err(RepoError::Msg(with_hint(format!(
525
                "{url}: HTTP {code}: {}",
526
                text.trim()
527
            ))))
528
        }
529
        // The transport error already names the URL.
530
        Err(e) => Err(RepoError::Msg(with_hint(e.to_string()))),
531
    }
532
}
533
534
impl Remote for HttpRemote {
535
    fn repo_id(&self) -> Result<[u8; 16]> {
536
        wire::decode_capabilities(&self.call(false, "capabilities", None)?)
537
    }
538
    fn refs(&self) -> Result<RefsResponse> {
539
        Ok(wire::decode_refs(&self.call(false, "refs", None)?)?.1)
540
    }
541
    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>> {
542
        wire::decode_hashes(&self.call(
543
            true,
544
            "negotiate",
545
            Some(&wire::encode_negotiate(have, want)),
546
        )?)
547
    }
548
    fn fetch(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle> {
549
        wire::decode_bundle(&self.call(
550
            true,
551
            "fetch",
552
            Some(&wire::encode_fetch(changes, max_blob)),
553
        )?)
554
    }
555
    fn objects(&self, refs: &[ObjectRef]) -> Result<Vec<(ObjectType, Vec<u8>)>> {
556
        Ok(
557
            wire::decode_bundle(&self.call(true, "objects", Some(&wire::encode_hashes(refs)))?)?
558
                .objects,
559
        )
560
    }
561
    fn checkpoint(&self, heads: &[ChangeHash]) -> Result<Bundle> {
562
        wire::decode_bundle(&self.call(true, "checkpoint", Some(&wire::encode_hashes(heads)))?)
563
    }
564
    fn shallow(&self, heads: &[ChangeHash], depth: u32) -> Result<Vec<ChangeHash>> {
565
        wire::decode_hashes(&self.call(
566
            true,
567
            "shallow",
568
            Some(&wire::encode_shallow(heads, depth)),
569
        )?)
570
    }
571
    fn push(
572
        &self,
573
        bundle: Bundle,
574
        updates: Vec<RefUpdate>,
575
        policy: Option<SignedPolicy>,
576
        tags: Vec<TagRef>,
577
    ) -> Result<Vec<RefResult>> {
578
        wire::decode_push_response(&self.call(
579
            true,
580
            "push",
581
            Some(&wire::encode_push(
582
                &bundle,
583
                &updates,
584
                policy.as_ref(),
585
                &tags,
586
            )),
587
        )?)
588
    }
589
}
590
591
/// A remote's URL as it is stored: an `http(s)://` URL as given, a local
592
/// path made absolute (with its `file://` kept), so it still names the same
593
/// repository from inside the clone (oxygen#57).
594
pub fn stored_url(url: &str) -> String {
595
    // Only a path is made absolute: a plain one, or one behind file:// or
596
    // dir://. Every other URL (a hub, a bucket) is kept as written.
597
    let (prefix, path) = match (url.strip_prefix("file://"), url.strip_prefix("dir://")) {
598
        (Some(p), _) => ("file://", p),
599
        (_, Some(p)) => ("dir://", p),
600
        _ if url.contains("://") => return url.to_string(),
601
        _ => ("", url),
602
    };
603
    match std::path::absolute(path) {
604
        Ok(abs) => format!("{prefix}{}", abs.display()),
605
        Err(_) => url.to_string(),
606
    }
607
}
608
609
/// Open a remote by URL: a local path (optionally `file://`), `http(s)://`
610
/// (a hub), or a store (`dir://`, and whatever openers are registered).
611
pub fn open_remote(url: &str, options: &RemoteOptions) -> Result<Box<dyn Remote>> {
612
    if url.starts_with("http://") || url.starts_with("https://") {
613
        return Ok(Box::new(HttpRemote::new(url, options)?));
614
    }
615
    if let Some(store) = open_store(url, options) {
616
        return crate::repo::strata::open_store_remote(store?, options);
617
    }
618
    let path = url.strip_prefix("file://").unwrap_or(url);
619
    if path.contains("://") {
620
        return Err(RepoError::Msg(format!("unsupported remote URL {url:?}")));
621
    }
622
    let mut local = LocalRemote::open(Path::new(path))?;
623
    local.claimed_id = options.repo_id;
624
    Ok(Box::new(local))
625
}
626
627
// ---------------------------------------------------------------- wire
628
629
/// CBOR encodings of the protocol messages (spec §16.1).
630
pub mod wire {
631
    use super::*;
632
633
    fn schema(msg: &str) -> RepoError {
634
        RepoError::Object(ObjectError::Schema(msg.into()))
635
    }
636
    fn sigs_value(sigs: &[Signature]) -> Value {
637
        Value::Array(sigs.iter().map(|s| s.to_value()).collect())
638
    }
639
    fn sigs_from(v: &Value) -> Result<Vec<Signature>> {
640
        Ok(o::array(v, "signatures")?
641
            .iter()
642
            .map(Signature::from_value)
643
            .collect::<std::result::Result<Vec<_>, _>>()?)
644
    }
645
    fn hashes_value(hs: &[ObjectRef]) -> Value {
646
        Value::Array(hs.iter().map(|h| Value::Bytes(h.0.to_vec())).collect())
647
    }
648
    fn hashes_from(v: &Value) -> Result<Vec<ObjectRef>> {
649
        Ok(o::array(v, "hashes")?
650
            .iter()
651
            .map(|h| o::oref(h, "hash"))
652
            .collect::<std::result::Result<Vec<_>, _>>()?)
653
    }
654
    fn attest_value(sa: &SignedAttest) -> Value {
655
        Value::Array(vec![Value::Bytes(sa.attest.encode()), sigs_value(&sa.sigs)])
656
    }
657
    fn attest_from(v: &Value) -> Result<SignedAttest> {
658
        let a = o::array(v, "signed attest")?;
659
        if a.len() != 2 {
660
            return Err(schema("signed attest: expected [attest, sigs]"));
661
        }
662
        Ok(SignedAttest {
663
            attest: Attest::decode(o::bytes(&a[0], "attest")?)?,
664
            sigs: sigs_from(&a[1])?,
665
        })
666
    }
667
    fn opt_attest_value(sa: &Option<SignedAttest>) -> Value {
668
        sa.as_ref().map(attest_value).unwrap_or(Value::Null)
669
    }
670
    fn opt_attest_from(v: &Value) -> Result<Option<SignedAttest>> {
671
        if v.is_null() {
672
            Ok(None)
673
        } else {
674
            attest_from(v).map(Some)
675
        }
676
    }
677
    /// `[policy bytes, [signature]]`: also the format of a policy proposal
678
    /// file exchanged between root holders (spec §14.5).
679
    pub fn policy_value(sp: &SignedPolicy) -> Value {
680
        Value::Array(vec![Value::Bytes(sp.policy.encode()), sigs_value(&sp.sigs)])
681
    }
682
    pub fn policy_from(v: &Value) -> Result<SignedPolicy> {
683
        let a = o::array(v, "signed policy")?;
684
        if a.len() != 2 {
685
            return Err(schema("signed policy: expected [policy, sigs]"));
686
        }
687
        Ok(SignedPolicy {
688
            policy: Policy::decode(o::bytes(&a[0], "policy")?)?,
689
            sigs: sigs_from(&a[1])?,
690
        })
691
    }
692
    fn opt_policy_value(sp: Option<&SignedPolicy>) -> Value {
693
        sp.map(policy_value).unwrap_or(Value::Null)
694
    }
695
    fn opt_policy_from(v: &Value) -> Result<Option<SignedPolicy>> {
696
        if v.is_null() {
697
            Ok(None)
698
        } else {
699
            policy_from(v).map(Some)
700
        }
701
    }
702
    fn map<'a>(v: &'a Value, what: &str) -> Result<&'a o::Map> {
703
        Ok(o::map(v, what)?)
704
    }
705
    fn field<'a>(m: &'a o::Map, k: u64, what: &str) -> Result<&'a Value> {
706
        Ok(o::field(m, k, what)?)
707
    }
708
709
    /// A bundle's value. Each change's text blobs and their chunks (spec
710
    /// §6.1) are added to its objects when they are not already there, so a
711
    /// bundle can always be read on its own.
712
    pub fn bundle_value(b: &Bundle) -> Value {
713
        let mut changes = Vec::with_capacity(b.raw_changes.len() + b.changes.len());
714
        // Changes as stored are already their encoding, their texts among
715
        // the objects.
716
        for (bytes, s) in &b.raw_changes {
717
            changes.push(Value::Array(vec![
718
                Value::Bytes(bytes.clone()),
719
                sigs_value(s),
720
            ]));
721
        }
722
        let mut texts: Vec<(ObjectType, Vec<u8>)> = Vec::new();
723
        let mut listed: Option<std::collections::HashSet<ObjectRef>> = None;
724
        for (c, s) in &b.changes {
725
            let (bytes, stored) = c.encode_parts();
726
            for t in stored {
727
                let listed = listed.get_or_insert_with(|| {
728
                    b.objects
729
                        .iter()
730
                        .map(|(ty, body)| crate::hash::hash_object(*ty, body))
731
                        .collect()
732
                });
733
                if !listed.insert(t.blob.hash()) {
734
                    continue;
735
                }
736
                for chunk in crate::chunk::chunks(&t.content) {
737
                    if listed.insert(crate::hash::hash_object(ObjectType::Chunk, chunk)) {
738
                        texts.push((ObjectType::Chunk, chunk.to_vec()));
739
                    }
740
                }
741
                texts.push((ObjectType::Blob, t.blob.encode()));
742
            }
743
            changes.push(Value::Array(vec![Value::Bytes(bytes), sigs_value(s)]));
744
        }
745
        let mut v = Value::map([
746
            (0, Value::Array(changes)),
747
            (
748
                1,
749
                Value::Array(b.attests.iter().map(attest_value).collect()),
750
            ),
751
            (
752
                2,
753
                Value::Array(
754
                    b.objects
755
                        .iter()
756
                        .chain(&texts)
757
                        .map(|(t, body)| {
758
                            Value::Array(vec![
759
                                Value::Text(t.name().into()),
760
                                Value::Bytes(body.clone()),
761
                            ])
762
                        })
763
                        .collect(),
764
                ),
765
            ),
766
            (
767
                3,
768
                Value::Array(b.policies.iter().map(policy_value).collect()),
769
            ),
770
        ]);
771
        // Only when there are any; older peers ignore the field.
772
        if !b.certificates.is_empty() {
773
            if let Value::Map(m) = &mut v {
774
                m.insert(
775
                    4,
776
                    Value::Array(
777
                        b.certificates
778
                            .iter()
779
                            .map(|c| Value::Bytes(c.encode()))
780
                            .collect(),
781
                    ),
782
                );
783
            }
784
        }
785
        v
786
    }
787
788
    pub fn bundle_from_value(v: &Value) -> Result<Bundle> {
789
        let m = map(v, "bundle")?;
790
        let mut b = Bundle::default();
791
        for ob in o::array(field(m, 2, "objects")?, "objects")? {
792
            let pair = o::array(ob, "object")?;
793
            if pair.len() != 2 {
794
                return Err(schema("object: expected [type, body]"));
795
            }
796
            let ty = ObjectType::from_name(o::text(&pair[0], "object type")?)
797
                .ok_or_else(|| schema("unknown object type"))?;
798
            b.objects
799
                .push((ty, o::bytes(&pair[1], "object body")?.to_vec()));
800
        }
801
        // Stored inserts are read from the bundle's own objects.
802
        let mut by_ref: Option<std::collections::HashMap<ObjectRef, usize>> = None;
803
        for c in o::array(field(m, 0, "changes")?, "changes")? {
804
            let pair = o::array(c, "change entry")?;
805
            if pair.len() != 2 {
806
                return Err(schema("change entry: expected [change, sigs]"));
807
            }
808
            let encoded = o::bytes(&pair[0], "change")?;
809
            b.hashes
810
                .push(crate::hash::hash_object(ObjectType::Change, encoded));
811
            let change = Change::decode_with(encoded, &mut |text, count, form| {
812
                let by_ref = by_ref.get_or_insert_with(|| {
813
                    b.objects
814
                        .iter()
815
                        .enumerate()
816
                        .map(|(i, (ty, body))| (crate::hash::hash_object(*ty, body), i))
817
                        .collect()
818
                });
819
                let get =
820
                    |r: &ObjectRef, want: ObjectType| match by_ref.get(r).map(|i| &b.objects[*i]) {
821
                        Some((ty, body)) if *ty == want => Ok(body),
822
                        _ => Err(ObjectError::Schema(format!(
823
                            "the bundle lacks {} {}",
824
                            want.name(),
825
                            r.short(12)
826
                        ))),
827
                    };
828
                let blob = crate::objects::Blob::decode(get(text, ObjectType::Blob)?)?;
829
                let mut content = Vec::with_capacity(blob.length as usize);
830
                for chunk in &blob.chunks {
831
                    content.extend_from_slice(get(chunk, ObjectType::Chunk)?);
832
                }
833
                crate::objects::text_lines(&content, count, form)
834
            })?;
835
            b.changes.push((change, sigs_from(&pair[1])?));
836
        }
837
        for a in o::array(field(m, 1, "attests")?, "attests")? {
838
            b.attests.push(attest_from(a)?);
839
        }
840
        if let Some(ps) = m.get(&3) {
841
            for p in o::array(ps, "policies")? {
842
                b.policies.push(policy_from(p)?);
843
            }
844
        }
845
        if let Some(cs) = m.get(&4) {
846
            for c in o::array(cs, "certificates")? {
847
                b.certificates.push(
848
                    crate::cert::SignedCertificate::decode(o::bytes(c, "certificate")?)
849
                        .map_err(|e| schema(&format!("certificate: {e}")))?,
850
                );
851
            }
852
        }
853
        Ok(b)
854
    }
855
856
    pub fn encode_bundle(b: &Bundle) -> Vec<u8> {
857
        cbor::encode(&bundle_value(b))
858
    }
859
    pub fn decode_bundle(bytes: &[u8]) -> Result<Bundle> {
860
        bundle_from_value(&cbor::decode(bytes).map_err(ObjectError::from)?)
861
    }
862
863
    pub fn encode_capabilities(id: [u8; 16]) -> Vec<u8> {
864
        cbor::encode(&Value::map([
865
            (0, Value::Array(vec![Value::Uint(PROTOCOL_VERSION)])),
866
            (1, Value::Bytes(id.to_vec())),
867
            (2, Value::Array(vec![Value::Text("seal".into())])),
868
        ]))
869
    }
870
    pub fn decode_capabilities(bytes: &[u8]) -> Result<[u8; 16]> {
871
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
872
        let m = map(&v, "capabilities")?;
873
        let versions = o::array(field(m, 0, "versions")?, "versions")?;
874
        if !versions
875
            .iter()
876
            .any(|x| x.as_uint() == Some(PROTOCOL_VERSION))
877
        {
878
            return Err(RepoError::Msg(
879
                "remote does not speak protocol version 1".into(),
880
            ));
881
        }
882
        o::bytes(field(m, 1, "repo id")?, "repo id")?
883
            .try_into()
884
            .map_err(|_| schema("repo id: expected 16 bytes"))
885
    }
886
887
    fn tags_value(tags: &[TagRef]) -> Value {
888
        Value::Array(
889
            tags.iter()
890
                .map(|t| {
891
                    Value::map([
892
                        (0, Value::Text(t.name.clone())),
893
                        (1, Value::Bytes(t.state.0.to_vec())),
894
                        (2, hashes_value(&t.heads)),
895
                        (3, opt_attest_value(&t.attestation)),
896
                    ])
897
                })
898
                .collect(),
899
        )
900
    }
901
    fn tags_from(v: &Value) -> Result<Vec<TagRef>> {
902
        let mut out = Vec::new();
903
        for t in o::array(v, "tags")? {
904
            let tm = map(t, "tag")?;
905
            out.push(TagRef {
906
                name: o::text(field(tm, 0, "name")?, "name")?.to_owned(),
907
                state: o::oref(field(tm, 1, "state")?, "state")?,
908
                heads: hashes_from(field(tm, 2, "heads")?)?,
909
                attestation: opt_attest_from(field(tm, 3, "attestation")?)?,
910
            });
911
        }
912
        Ok(out)
913
    }
914
915
    pub fn encode_refs(id: [u8; 16], resp: &RefsResponse) -> Vec<u8> {
916
        cbor::encode(&Value::map([
917
            (
918
                0,
919
                Value::Array(
920
                    resp.refs
921
                        .iter()
922
                        .map(|r| {
923
                            let mut v = Value::map([
924
                                (0, Value::Text(r.name.clone())),
925
                                (1, Value::Bytes(r.state.0.to_vec())),
926
                                (2, hashes_value(&r.heads)),
927
                                (3, opt_attest_value(&r.attestation)),
928
                                (4, r.last_serial.map(Value::Uint).unwrap_or(Value::Null)),
929
                            ]);
930
                            // Only when there is one; older peers ignore it.
931
                            if r.accepted.is_some() {
932
                                if let Value::Map(m) = &mut v {
933
                                    m.insert(5, opt_attest_value(&r.accepted));
934
                                }
935
                            }
936
                            v
937
                        })
938
                        .collect(),
939
                ),
940
            ),
941
            (1, Value::Bytes(id.to_vec())),
942
            (2, opt_policy_value(resp.policy.as_ref())),
943
            (3, tags_value(&resp.tags)),
944
            (
945
                4,
946
                Value::Array(resp.history.iter().map(policy_value).collect()),
947
            ),
948
            (
949
                5,
950
                resp.default_bookmark
951
                    .clone()
952
                    .map(Value::Text)
953
                    .unwrap_or(Value::Null),
954
            ),
955
        ]))
956
    }
957
    pub fn decode_refs(bytes: &[u8]) -> Result<([u8; 16], RefsResponse)> {
958
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
959
        let m = map(&v, "refs")?;
960
        let id: [u8; 16] = o::bytes(field(m, 1, "repo id")?, "repo id")?
961
            .try_into()
962
            .map_err(|_| schema("repo id: expected 16 bytes"))?;
963
        let mut refs = Vec::new();
964
        for r in o::array(field(m, 0, "refs")?, "refs")? {
965
            let rm = map(r, "ref")?;
966
            let ls = field(rm, 4, "last serial")?;
967
            refs.push(RemoteRef {
968
                name: o::text(field(rm, 0, "name")?, "name")?.to_owned(),
969
                state: o::oref(field(rm, 1, "state")?, "state")?,
970
                heads: hashes_from(field(rm, 2, "heads")?)?,
971
                attestation: opt_attest_from(field(rm, 3, "attestation")?)?,
972
                last_serial: if ls.is_null() {
973
                    None
974
                } else {
975
                    Some(o::uint(ls, "last serial")?)
976
                },
977
                accepted: match rm.get(&5) {
978
                    Some(a) => opt_attest_from(a)?,
979
                    None => None,
980
                },
981
            });
982
        }
983
        let policy = match m.get(&2) {
984
            Some(p) => opt_policy_from(p)?,
985
            None => None,
986
        };
987
        let tags = match m.get(&3) {
988
            Some(t) => tags_from(t)?,
989
            None => Vec::new(),
990
        };
991
        let mut history = Vec::new();
992
        if let Some(h) = m.get(&4) {
993
            for p in o::array(h, "history")? {
994
                history.push(policy_from(p)?);
995
            }
996
        }
997
        let default_bookmark = match m.get(&5) {
998
            Some(Value::Text(t)) => Some(t.clone()),
999
            _ => None,
1000
        };
1001
        Ok((
1002
            id,
1003
            RefsResponse {
1004
                refs,
1005
                policy,
1006
                tags,
1007
                history,
1008
                default_bookmark,
1009
            },
1010
        ))
1011
    }
1012
1013
    pub fn encode_negotiate(have: &[ChangeHash], want: &[ChangeHash]) -> Vec<u8> {
1014
        cbor::encode(&Value::map([
1015
            (0, hashes_value(have)),
1016
            (1, hashes_value(want)),
1017
        ]))
1018
    }
1019
    pub fn decode_negotiate(bytes: &[u8]) -> Result<(Vec<ChangeHash>, Vec<ChangeHash>)> {
1020
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
1021
        let m = map(&v, "negotiate")?;
1022
        Ok((
1023
            hashes_from(field(m, 0, "have")?)?,
1024
            hashes_from(field(m, 1, "want")?)?,
1025
        ))
1026
    }
1027
1028
    /// `fetch` request: `{0: [changes], 1: max blob size or null}`.
1029
    pub fn encode_fetch(hs: &[ChangeHash], max_blob: Option<u64>) -> Vec<u8> {
1030
        cbor::encode(&Value::map([
1031
            (0, hashes_value(hs)),
1032
            (1, max_blob.map(Value::Uint).unwrap_or(Value::Null)),
1033
        ]))
1034
    }
1035
    pub fn decode_fetch(bytes: &[u8]) -> Result<(Vec<ChangeHash>, Option<u64>)> {
1036
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
1037
        let m = map(&v, "fetch")?;
1038
        let max = match m.get(&1) {
1039
            Some(x) if !x.is_null() => Some(o::uint(x, "max blob")?),
1040
            _ => None,
1041
        };
1042
        Ok((hashes_from(field(m, 0, "changes")?)?, max))
1043
    }
1044
1045
    /// `shallow` request: `{0: [heads], 1: depth}`.
1046
    pub fn encode_shallow(hs: &[ChangeHash], depth: u32) -> Vec<u8> {
1047
        cbor::encode(&Value::map([
1048
            (0, hashes_value(hs)),
1049
            (1, Value::Uint(depth as u64)),
1050
        ]))
1051
    }
1052
    pub fn decode_shallow(bytes: &[u8]) -> Result<(Vec<ChangeHash>, u32)> {
1053
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
1054
        let m = map(&v, "shallow")?;
1055
        Ok((
1056
            hashes_from(field(m, 0, "heads")?)?,
1057
            o::u32_(field(m, 1, "depth")?, "depth")?,
1058
        ))
1059
    }
1060
1061
    pub fn encode_hashes(hs: &[ChangeHash]) -> Vec<u8> {
1062
        cbor::encode(&Value::map([(0, hashes_value(hs))]))
1063
    }
1064
    pub fn decode_hashes(bytes: &[u8]) -> Result<Vec<ChangeHash>> {
1065
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
1066
        hashes_from(field(map(&v, "hashes")?, 0, "hashes")?)
1067
    }
1068
1069
    pub fn encode_push(
1070
        bundle: &Bundle,
1071
        updates: &[RefUpdate],
1072
        policy: Option<&SignedPolicy>,
1073
        tags: &[TagRef],
1074
    ) -> Vec<u8> {
1075
        cbor::encode(&Value::map([
1076
            (0, bundle_value(bundle)),
1077
            (2, opt_policy_value(policy)),
1078
            (3, tags_value(tags)),
1079
            (
1080
                1,
1081
                Value::Array(
1082
                    updates
1083
                        .iter()
1084
                        .map(|u| {
1085
                            Value::map([
1086
                                (0, Value::Text(u.name.clone())),
1087
                                (1, Value::Bytes(u.old_state.0.to_vec())),
1088
                                (2, Value::Bytes(u.new_state.0.to_vec())),
1089
                                (3, Value::Bool(u.force)),
1090
                                (4, opt_attest_value(&u.attestation)),
1091
                            ])
1092
                        })
1093
                        .collect(),
1094
                ),
1095
            ),
1096
        ]))
1097
    }
1098
    /// A decoded push: bundle, bookmark updates, policy, tags.
1099
    pub type PushParts = (Bundle, Vec<RefUpdate>, Option<SignedPolicy>, Vec<TagRef>);
1100
1101
    pub fn decode_push(bytes: &[u8]) -> Result<PushParts> {
1102
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
1103
        let m = map(&v, "push")?;
1104
        let bundle = bundle_from_value(field(m, 0, "bundle")?)?;
1105
        let mut updates = Vec::new();
1106
        for u in o::array(field(m, 1, "updates")?, "updates")? {
1107
            let um = map(u, "update")?;
1108
            updates.push(RefUpdate {
1109
                name: o::text(field(um, 0, "name")?, "name")?.to_owned(),
1110
                old_state: o::oref(field(um, 1, "old_state")?, "old_state")?,
1111
                new_state: o::oref(field(um, 2, "new_state")?, "new_state")?,
1112
                force: field(um, 3, "force")?
1113
                    .as_bool()
1114
                    .ok_or_else(|| schema("force: expected bool"))?,
1115
                attestation: opt_attest_from(field(um, 4, "attestation")?)?,
1116
            });
1117
        }
1118
        let policy = match m.get(&2) {
1119
            Some(p) => opt_policy_from(p)?,
1120
            None => None,
1121
        };
1122
        let tags = match m.get(&3) {
1123
            Some(t) => tags_from(t)?,
1124
            None => Vec::new(),
1125
        };
1126
        Ok((bundle, updates, policy, tags))
1127
    }
1128
1129
    pub fn encode_push_response(results: &[RefResult]) -> Vec<u8> {
1130
        cbor::encode(&Value::map([(
1131
            0,
1132
            Value::Array(
1133
                results
1134
                    .iter()
1135
                    .map(|r| {
1136
                        Value::map([
1137
                            (0, Value::Text(r.name.clone())),
1138
                            (1, Value::Bool(r.ok)),
1139
                            (2, Value::Text(r.reason.clone())),
1140
                        ])
1141
                    })
1142
                    .collect(),
1143
            ),
1144
        )]))
1145
    }
1146
    pub fn decode_push_response(bytes: &[u8]) -> Result<Vec<RefResult>> {
1147
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
1148
        let m = map(&v, "push response")?;
1149
        let mut out = Vec::new();
1150
        for r in o::array(field(m, 0, "results")?, "results")? {
1151
            let rm = map(r, "result")?;
1152
            out.push(RefResult {
1153
                name: o::text(field(rm, 0, "name")?, "name")?.to_owned(),
1154
                ok: field(rm, 1, "ok")?
1155
                    .as_bool()
1156
                    .ok_or_else(|| schema("ok: expected bool"))?,
1157
                reason: o::text(field(rm, 2, "reason")?, "reason")?.to_owned(),
1158
            });
1159
        }
1160
        Ok(out)
1161
    }
1162
}
1163
1164
#[cfg(test)]
1165
mod tests {
1166
    use super::*;
1167
    use crate::attest::AttestBody;
1168
    use crate::objects::Hlc;
1169
1170
    #[test]
1171
    fn wire_round_trips() {
1172
        let c = crate::objects::tests::sample_change();
1173
        let sk = crate::identity::SecretKey {
1174
            ed_seed: [1; 32],
1175
            ml_seed: [2; 32],
1176
        };
1177
        let id = sk.identity("a", "b", 0);
1178
        let key = id.hash();
1179
        let sig = sk.sign(crate::identity::Domain::Change, &key, &c.hash());
1180
        let seal = Attest::seal([3; 16], vec![c.hash()]);
1181
        let seal_sig = sk.sign(crate::identity::Domain::Attest, &key, &seal.hash());
1182
        let policy = Policy::single_root([3; 16], id.clone(), 99);
1183
        let psig = sk.sign(crate::identity::Domain::Policy, &key, &policy.hash());
1184
        let bundle = Bundle {
1185
            changes: vec![(c.clone(), vec![sig.clone()])],
1186
            attests: vec![SignedAttest {
1187
                attest: seal.clone(),
1188
                sigs: vec![seal_sig],
1189
            }],
1190
            objects: vec![
1191
                (ObjectType::Identity, id.encode()),
1192
                (ObjectType::Chunk, b"xyz".to_vec()),
1193
            ],
1194
            policies: vec![SignedPolicy {
1195
                policy: policy.clone(),
1196
                sigs: vec![psig.clone()],
1197
            }],
1198
            hashes: vec![],
1199
            raw_changes: vec![],
1200
            certificates: vec![],
1201
        };
1202
        let d = wire::decode_bundle(&wire::encode_bundle(&bundle)).unwrap();
1203
        assert_eq!(
1204
            d.hashes,
1205
            vec![c.hash()],
1206
            "the receiver hashes what it was sent"
1207
        );
1208
        assert_eq!(d.changes, bundle.changes);
1209
        assert_eq!(d.attests, bundle.attests);
1210
        assert_eq!(d.objects, bundle.objects);
1211
        assert_eq!(d.policies, bundle.policies);
1212
        assert!(d.certificates.is_empty());
1213
        // Certificates (keyless, oxygenhub#241) travel as field 4, only
1214
        // when there are any: a bundle without them encodes as before.
1215
        let plain = wire::encode_bundle(&bundle);
1216
        let mut with = bundle.clone();
1217
        let issuer_sk = crate::identity::SecretKey::generate();
1218
        let issuer = issuer_sk.identity("hub", "h@x", 1);
1219
        with.certificates.push(crate::cert::SignedCertificate::sign(
1220
            crate::cert::Certificate {
1221
                key: id.clone(),
1222
                issuer: issuer.hash(),
1223
                subject: "agent:a@b".into(),
1224
                claims: vec![],
1225
                scope: crate::identity::Scope::CHANGES,
1226
                not_before: 1,
1227
                expires: 2,
1228
                grant: None,
1229
            },
1230
            &issuer,
1231
            &issuer_sk,
1232
        ));
1233
        let encoded = wire::encode_bundle(&with);
1234
        assert_ne!(encoded, plain);
1235
        assert_eq!(
1236
            wire::decode_bundle(&encoded).unwrap().certificates,
1237
            with.certificates
1238
        );
1239
        with.certificates.clear();
1240
        assert_eq!(wire::encode_bundle(&with), plain);
1241
1242
        let att = Attest {
1243
            repo: [3; 16],
1244
            hlc: Hlc::default(),
1245
            body: AttestBody::Ref {
1246
                name: "main".into(),
1247
                old_state: ObjectRef([1; 32]),
1248
                new_state: ObjectRef([2; 32]),
1249
                serial: 4,
1250
            },
1251
        };
1252
        let refs = vec![
1253
            RemoteRef {
1254
                name: "main".into(),
1255
                state: ObjectRef([2; 32]),
1256
                heads: vec![c.hash()],
1257
                attestation: Some(SignedAttest {
1258
                    attest: att.clone(),
1259
                    sigs: vec![sig.clone()],
1260
                }),
1261
                last_serial: Some(4),
1262
                accepted: None,
1263
            },
1264
            RemoteRef {
1265
                name: "dev".into(),
1266
                state: ObjectRef([5; 32]),
1267
                heads: vec![],
1268
                attestation: None,
1269
                last_serial: None,
1270
                accepted: None,
1271
            },
1272
        ];
1273
        let tag = TagRef {
1274
            name: "v1".into(),
1275
            state: ObjectRef([2; 32]),
1276
            heads: vec![c.hash()],
1277
            attestation: None,
1278
        };
1279
        let resp = RefsResponse {
1280
            refs,
1281
            policy: Some(SignedPolicy {
1282
                policy: policy.clone(),
1283
                sigs: vec![psig.clone()],
1284
            }),
1285
            tags: vec![tag.clone()],
1286
            history: vec![SignedPolicy {
1287
                policy: policy.clone(),
1288
                sigs: vec![],
1289
            }],
1290
            default_bookmark: Some("develop".into()),
1291
        };
1292
        let (id_back, back) = wire::decode_refs(&wire::encode_refs([3; 16], &resp)).unwrap();
1293
        assert_eq!(id_back, [3; 16]);
1294
        assert_eq!(back.refs.len(), 2);
1295
        assert_eq!(back.refs[0].attestation.as_ref().unwrap().attest, att);
1296
        assert_eq!(back.refs[1].last_serial, None);
1297
        assert_eq!(back.policy.as_ref().unwrap().policy, policy);
1298
        assert_eq!(back.history, resp.history);
1299
        assert_eq!(back.tags, vec![tag.clone()]);
1300
1301
        let (h, w) = wire::decode_negotiate(&wire::encode_negotiate(
1302
            &[ObjectRef([1; 32])],
1303
            &[ObjectRef([2; 32])],
1304
        ))
1305
        .unwrap();
1306
        assert_eq!((h, w), (vec![ObjectRef([1; 32])], vec![ObjectRef([2; 32])]));
1307
        assert_eq!(
1308
            wire::decode_hashes(&wire::encode_hashes(&[c.hash()])).unwrap(),
1309
            vec![c.hash()]
1310
        );
1311
        assert_eq!(
1312
            wire::decode_fetch(&wire::encode_fetch(&[c.hash()], Some(4096))).unwrap(),
1313
            (vec![c.hash()], Some(4096))
1314
        );
1315
        assert_eq!(
1316
            wire::decode_fetch(&wire::encode_fetch(&[], None)).unwrap(),
1317
            (vec![], None)
1318
        );
1319
        assert_eq!(
1320
            wire::decode_shallow(&wire::encode_shallow(&[c.hash()], 3)).unwrap(),
1321
            (vec![c.hash()], 3)
1322
        );
1323
        // An old-style fetch request without the filter field still decodes.
1324
        assert_eq!(
1325
            wire::decode_fetch(&wire::encode_hashes(&[c.hash()])).unwrap(),
1326
            (vec![c.hash()], None)
1327
        );
1328
1329
        let up = RefUpdate {
1330
            name: "main".into(),
1331
            old_state: ObjectRef([1; 32]),
1332
            new_state: ObjectRef([2; 32]),
1333
            force: true,
1334
            attestation: Some(SignedAttest {
1335
                attest: att,
1336
                sigs: vec![sig],
1337
            }),
1338
        };
1339
        let sp = SignedPolicy {
1340
            policy,
1341
            sigs: vec![psig],
1342
        };
1343
        let (b2, ups, pol, tags) = wire::decode_push(&wire::encode_push(
1344
            &bundle,
1345
            std::slice::from_ref(&up),
1346
            Some(&sp),
1347
            std::slice::from_ref(&tag),
1348
        ))
1349
        .unwrap();
1350
        assert_eq!(pol, Some(sp));
1351
        assert_eq!(tags, vec![tag]);
1352
        assert_eq!(b2.changes.len(), 1);
1353
        assert_eq!(ups[0].name, "main");
1354
        assert!(ups[0].force);
1355
        assert_eq!(ups[0].attestation, up.attestation);
1356
        let res = vec![RefResult {
1357
            name: "main".into(),
1358
            ok: false,
1359
            reason: "stale".into(),
1360
        }];
1361
        assert_eq!(
1362
            wire::decode_push_response(&wire::encode_push_response(&res)).unwrap(),
1363
            res
1364
        );
1365
        assert_eq!(
1366
            wire::decode_capabilities(&wire::encode_capabilities([7; 16])).unwrap(),
1367
            [7; 16]
1368
        );
1369
    }
1370
}