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