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