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