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