OXYGENHUB
# abaa8d517631 oxygen / crates / oxy-core / src / protocol.rs
18.9 KiB · Rust · last changed HTTP transport: CBOR wire messages, HttpRemote client, oxy-server crate with bearer-token pushes, oxy serve 1 month 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::repo::{Repo, RepoError, Result};
15 
use std::path::Path;
16
17 
pub const CONTENT_TYPE: &str = "application/oxy-cbor";
18 
pub const PROTOCOL_VERSION: u64 = 1;
19
20 
/// An attestation together with the signatures over its ObjectRef.
21 
#[derive(Clone, Debug, PartialEq, Eq)]
22 
pub struct SignedAttest {
23 
    pub attest: Attest,
24 
    pub sigs: Vec<Signature>,
25 
}
26
27 
/// One reference as a remote reports it.
28 
#[derive(Clone, Debug)]
29 
pub struct RemoteRef {
30 
    pub name: String,
31 
    pub state: ObjectRef,
32 
    pub heads: Vec<ChangeHash>,
33 
    pub attestation: Option<SignedAttest>,
34 
}
35
36 
/// Everything needed to add a set of changes to a repository.
37 
#[derive(Clone, Debug, Default)]
38 
pub struct Bundle {
39 
    /// In dependency order, each with its individual signatures.
40 
    pub changes: Vec<(Change, Vec<Signature>)>,
41 
    /// Seals covering the changes.
42 
    pub attests: Vec<SignedAttest>,
43 
    /// Chunks, blobs, identities, and state objects referenced by the changes.
44 
    pub objects: Vec<(ObjectType, Vec<u8>)>,
45 
}
46
47 
impl Bundle {
48 
    pub fn is_empty(&self) -> bool {
49 
        self.changes.is_empty() && self.attests.is_empty() && self.objects.is_empty()
50 
    }
51 
}
52
53 
/// A requested bookmark move (spec §10.2, §16.3).
54 
#[derive(Clone, Debug)]
55 
pub struct RefUpdate {
56 
    pub name: String,
57 
    pub old_state: ObjectRef,
58 
    pub new_state: ObjectRef,
59 
    pub force: bool,
60 
    pub attestation: Option<SignedAttest>,
61 
}
62
63 
#[derive(Clone, Debug, PartialEq, Eq)]
64 
pub struct RefResult {
65 
    pub name: String,
66 
    pub ok: bool,
67 
    pub reason: String,
68 
}
69
70 
pub trait Remote {
71 
    fn repo_id(&self) -> Result<[u8; 16]>;
72 
    fn refs(&self) -> Result<Vec<RemoteRef>>;
73 
    /// Changes in the closure of `want` but not of `have`, in dependency order.
74 
    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>>;
75 
    fn fetch(&self, changes: &[ChangeHash]) -> Result<Bundle>;
76 
    fn push(&self, bundle: Bundle, updates: Vec<RefUpdate>) -> Result<Vec<RefResult>>;
77 
}
78
79 
// ---------------------------------------------------------------- local
80
81 
/// Another repository on this machine, served in-process.
82 
pub struct LocalRemote {
83 
    pub repo: Repo,
84 
}
85
86 
impl LocalRemote {
87 
    pub fn open(path: &Path) -> Result<LocalRemote> {
88 
        Ok(LocalRemote { repo: Repo::open(path)? })
89 
    }
90 
}
91
92 
impl Remote for LocalRemote {
93 
    fn repo_id(&self) -> Result<[u8; 16]> {
94 
        self.repo.id()
95 
    }
96 
    fn refs(&self) -> Result<Vec<RemoteRef>> {
97 
        self.repo.remote_refs()
98 
    }
99 
    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>> {
100 
        self.repo.negotiate(have, want)
101 
    }
102 
    fn fetch(&self, changes: &[ChangeHash]) -> Result<Bundle> {
103 
        self.repo.bundle_for(changes)
104 
    }
105 
    fn push(&self, bundle: Bundle, updates: Vec<RefUpdate>) -> Result<Vec<RefResult>> {
106 
        self.repo.receive_push(bundle, updates)
107 
    }
108 
}
109
110 
// ---------------------------------------------------------------- http
111
112 
/// A remote served by `oxy-server` at `<base>/oxy/v1/...`.
113 
pub struct HttpRemote {
114 
    base: String,
115 
    token: Option<String>,
116 
    agent: ureq::Agent,
117 
}
118
119 
impl HttpRemote {
120 
    pub fn new(base: &str, token: Option<String>) -> HttpRemote {
121 
        HttpRemote { base: base.trim_end_matches('/').to_string(), token, agent: ureq::Agent::new() }
122 
    }
123
124 
    fn call(&self, post: bool, endpoint: &str, body: Option<&[u8]>) -> Result<Vec<u8>> {
125 
        let url = format!("{}/oxy/v1/{endpoint}", self.base);
126 
        let mut req = if post { self.agent.post(&url) } else { self.agent.get(&url) };
127 
        req = req.set("Accept", CONTENT_TYPE);
128 
        if let Some(t) = &self.token {
129 
            req = req.set("Authorization", &format!("Bearer {t}"));
130 
        }
131 
        let resp = match body {
132 
            Some(b) => req.set("Content-Type", CONTENT_TYPE).send_bytes(b),
133 
            None => req.call(),
134 
        };
135 
        match resp {
136 
            Ok(r) => {
137 
                let mut out = Vec::new();
138 
                r.into_reader().read_to_end(&mut out).map_err(|e| RepoError::Msg(format!("{url}: {e}")))?;
139 
                Ok(out)
140 
            }
141 
            Err(ureq::Error::Status(code, r)) => {
142 
                let text = r.into_string().unwrap_or_default();
143 
                Err(RepoError::Msg(format!("{url}: HTTP {code}: {}", text.trim())))
144 
            }
145 
            Err(e) => Err(RepoError::Msg(format!("{url}: {e}"))),
146 
        }
147 
    }
148 
}
149
150 
impl Remote for HttpRemote {
151 
    fn repo_id(&self) -> Result<[u8; 16]> {
152 
        wire::decode_capabilities(&self.call(false, "capabilities", None)?)
153 
    }
154 
    fn refs(&self) -> Result<Vec<RemoteRef>> {
155 
        Ok(wire::decode_refs(&self.call(false, "refs", None)?)?.1)
156 
    }
157 
    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>> {
158 
        wire::decode_hashes(&self.call(true, "negotiate", Some(&wire::encode_negotiate(have, want)))?)
159 
    }
160 
    fn fetch(&self, changes: &[ChangeHash]) -> Result<Bundle> {
161 
        wire::decode_bundle(&self.call(true, "fetch", Some(&wire::encode_hashes(changes)))?)
162 
    }
163 
    fn push(&self, bundle: Bundle, updates: Vec<RefUpdate>) -> Result<Vec<RefResult>> {
164 
        wire::decode_push_response(&self.call(true, "push", Some(&wire::encode_push(&bundle, &updates)))?)
165 
    }
166 
}
167
168 
/// Open a remote by URL: a local path (optionally `file://`), or `http(s)://`.
169 
pub fn open_remote(url: &str, token: Option<String>) -> Result<Box<dyn Remote>> {
170 
    if url.starts_with("http://") || url.starts_with("https://") {
171 
        return Ok(Box::new(HttpRemote::new(url, token)));
172 
    }
173 
    let path = url.strip_prefix("file://").unwrap_or(url);
174 
    if path.contains("://") {
175 
        return Err(RepoError::Msg(format!("unsupported remote URL {url:?}")));
176 
    }
177 
    Ok(Box::new(LocalRemote::open(Path::new(path))?))
178 
}
179
180 
// ---------------------------------------------------------------- wire
181
182 
/// CBOR encodings of the protocol messages (spec §16.1).
183 
pub mod wire {
184 
    use super::*;
185
186 
    fn schema(msg: &str) -> RepoError {
187 
        RepoError::Object(ObjectError::Schema(msg.into()))
188 
    }
189 
    fn sigs_value(sigs: &[Signature]) -> Value {
190 
        Value::Array(sigs.iter().map(|s| s.to_value()).collect())
191 
    }
192 
    fn sigs_from(v: &Value) -> Result<Vec<Signature>> {
193 
        Ok(o::array(v, "signatures")?.iter().map(Signature::from_value).collect::<std::result::Result<Vec<_>, _>>()?)
194 
    }
195 
    fn hashes_value(hs: &[ObjectRef]) -> Value {
196 
        Value::Array(hs.iter().map(|h| Value::Bytes(h.0.to_vec())).collect())
197 
    }
198 
    fn hashes_from(v: &Value) -> Result<Vec<ObjectRef>> {
199 
        Ok(o::array(v, "hashes")?.iter().map(|h| o::oref(h, "hash")).collect::<std::result::Result<Vec<_>, _>>()?)
200 
    }
201 
    fn attest_value(sa: &SignedAttest) -> Value {
202 
        Value::Array(vec![Value::Bytes(sa.attest.encode()), sigs_value(&sa.sigs)])
203 
    }
204 
    fn attest_from(v: &Value) -> Result<SignedAttest> {
205 
        let a = o::array(v, "signed attest")?;
206 
        if a.len() != 2 {
207 
            return Err(schema("signed attest: expected [attest, sigs]"));
208 
        }
209 
        Ok(SignedAttest { attest: Attest::decode(o::bytes(&a[0], "attest")?)?, sigs: sigs_from(&a[1])? })
210 
    }
211 
    fn opt_attest_value(sa: &Option<SignedAttest>) -> Value {
212 
        sa.as_ref().map(attest_value).unwrap_or(Value::Null)
213 
    }
214 
    fn opt_attest_from(v: &Value) -> Result<Option<SignedAttest>> {
215 
        if v.is_null() { Ok(None) } else { attest_from(v).map(Some) }
216 
    }
217 
    fn map<'a>(v: &'a Value, what: &str) -> Result<&'a o::Map> {
218 
        Ok(o::map(v, what)?)
219 
    }
220 
    fn field<'a>(m: &'a o::Map, k: u64, what: &str) -> Result<&'a Value> {
221 
        Ok(o::field(m, k, what)?)
222 
    }
223
224 
    pub fn bundle_value(b: &Bundle) -> Value {
225 
        Value::map([
226 
            (
227 
                0,
228 
                Value::Array(b.changes.iter().map(|(c, s)| Value::Array(vec![Value::Bytes(c.encode()), sigs_value(s)])).collect()),
229 
            ),
230 
            (1, Value::Array(b.attests.iter().map(attest_value).collect())),
231 
            (
232 
                2,
233 
                Value::Array(
234 
                    b.objects.iter().map(|(t, body)| Value::Array(vec![Value::Text(t.name().into()), Value::Bytes(body.clone())])).collect(),
235 
                ),
236 
            ),
237 
        ])
238 
    }
239
240 
    pub fn bundle_from_value(v: &Value) -> Result<Bundle> {
241 
        let m = map(v, "bundle")?;
242 
        let mut b = Bundle::default();
243 
        for c in o::array(field(m, 0, "changes")?, "changes")? {
244 
            let pair = o::array(c, "change entry")?;
245 
            if pair.len() != 2 {
246 
                return Err(schema("change entry: expected [change, sigs]"));
247 
            }
248 
            b.changes.push((Change::decode(o::bytes(&pair[0], "change")?)?, sigs_from(&pair[1])?));
249 
        }
250 
        for a in o::array(field(m, 1, "attests")?, "attests")? {
251 
            b.attests.push(attest_from(a)?);
252 
        }
253 
        for ob in o::array(field(m, 2, "objects")?, "objects")? {
254 
            let pair = o::array(ob, "object")?;
255 
            if pair.len() != 2 {
256 
                return Err(schema("object: expected [type, body]"));
257 
            }
258 
            let ty = ObjectType::from_name(o::text(&pair[0], "object type")?).ok_or_else(|| schema("unknown object type"))?;
259 
            b.objects.push((ty, o::bytes(&pair[1], "object body")?.to_vec()));
260 
        }
261 
        Ok(b)
262 
    }
263
264 
    pub fn encode_bundle(b: &Bundle) -> Vec<u8> {
265 
        cbor::encode(&bundle_value(b))
266 
    }
267 
    pub fn decode_bundle(bytes: &[u8]) -> Result<Bundle> {
268 
        bundle_from_value(&cbor::decode(bytes).map_err(ObjectError::from)?)
269 
    }
270
271 
    pub fn encode_capabilities(id: [u8; 16]) -> Vec<u8> {
272 
        cbor::encode(&Value::map([
273 
            (0, Value::Array(vec![Value::Uint(PROTOCOL_VERSION)])),
274 
            (1, Value::Bytes(id.to_vec())),
275 
            (2, Value::Array(vec![Value::Text("seal".into())])),
276 
        ]))
277 
    }
278 
    pub fn decode_capabilities(bytes: &[u8]) -> Result<[u8; 16]> {
279 
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
280 
        let m = map(&v, "capabilities")?;
281 
        let versions = o::array(field(m, 0, "versions")?, "versions")?;
282 
        if !versions.iter().any(|x| x.as_uint() == Some(PROTOCOL_VERSION)) {
283 
            return Err(RepoError::Msg("remote does not speak protocol version 1".into()));
284 
        }
285 
        o::bytes(field(m, 1, "repo id")?, "repo id")?.try_into().map_err(|_| schema("repo id: expected 16 bytes"))
286 
    }
287
288 
    pub fn encode_refs(id: [u8; 16], refs: &[RemoteRef]) -> Vec<u8> {
289 
        cbor::encode(&Value::map([
290 
            (
291 
                0,
292 
                Value::Array(
293 
                    refs.iter()
294 
                        .map(|r| {
295 
                            Value::map([
296 
                                (0, Value::Text(r.name.clone())),
297 
                                (1, Value::Bytes(r.state.0.to_vec())),
298 
                                (2, hashes_value(&r.heads)),
299 
                                (3, opt_attest_value(&r.attestation)),
300 
                                (4, r.last_serial.map(Value::Uint).unwrap_or(Value::Null)),
301 
                            ])
302 
                        })
303 
                        .collect(),
304 
                ),
305 
            ),
306 
            (1, Value::Bytes(id.to_vec())),
307 
        ]))
308 
    }
309 
    pub fn decode_refs(bytes: &[u8]) -> Result<([u8; 16], Vec<RemoteRef>)> {
310 
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
311 
        let m = map(&v, "refs")?;
312 
        let id: [u8; 16] = o::bytes(field(m, 1, "repo id")?, "repo id")?.try_into().map_err(|_| schema("repo id: expected 16 bytes"))?;
313 
        let mut refs = Vec::new();
314 
        for r in o::array(field(m, 0, "refs")?, "refs")? {
315 
            let rm = map(r, "ref")?;
316 
            let ls = field(rm, 4, "last serial")?;
317 
            refs.push(RemoteRef {
318 
                name: o::text(field(rm, 0, "name")?, "name")?.to_owned(),
319 
                state: o::oref(field(rm, 1, "state")?, "state")?,
320 
                heads: hashes_from(field(rm, 2, "heads")?)?,
321 
                attestation: opt_attest_from(field(rm, 3, "attestation")?)?,
322 
                last_serial: if ls.is_null() { None } else { Some(o::uint(ls, "last serial")?) },
323 
            });
324 
        }
325 
        Ok((id, refs))
326 
    }
327
328 
    pub fn encode_negotiate(have: &[ChangeHash], want: &[ChangeHash]) -> Vec<u8> {
329 
        cbor::encode(&Value::map([(0, hashes_value(have)), (1, hashes_value(want))]))
330 
    }
331 
    pub fn decode_negotiate(bytes: &[u8]) -> Result<(Vec<ChangeHash>, Vec<ChangeHash>)> {
332 
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
333 
        let m = map(&v, "negotiate")?;
334 
        Ok((hashes_from(field(m, 0, "have")?)?, hashes_from(field(m, 1, "want")?)?))
335 
    }
336
337 
    pub fn encode_hashes(hs: &[ChangeHash]) -> Vec<u8> {
338 
        cbor::encode(&Value::map([(0, hashes_value(hs))]))
339 
    }
340 
    pub fn decode_hashes(bytes: &[u8]) -> Result<Vec<ChangeHash>> {
341 
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
342 
        hashes_from(field(map(&v, "hashes")?, 0, "hashes")?)
343 
    }
344
345 
    pub fn encode_push(bundle: &Bundle, updates: &[RefUpdate]) -> Vec<u8> {
346 
        cbor::encode(&Value::map([
347 
            (0, bundle_value(bundle)),
348 
            (
349 
                1,
350 
                Value::Array(
351 
                    updates
352 
                        .iter()
353 
                        .map(|u| {
354 
                            Value::map([
355 
                                (0, Value::Text(u.name.clone())),
356 
                                (1, Value::Bytes(u.old_state.0.to_vec())),
357 
                                (2, Value::Bytes(u.new_state.0.to_vec())),
358 
                                (3, Value::Bool(u.force)),
359 
                                (4, opt_attest_value(&u.attestation)),
360 
                            ])
361 
                        })
362 
                        .collect(),
363 
                ),
364 
            ),
365 
        ]))
366 
    }
367 
    pub fn decode_push(bytes: &[u8]) -> Result<(Bundle, Vec<RefUpdate>)> {
368 
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
369 
        let m = map(&v, "push")?;
370 
        let bundle = bundle_from_value(field(m, 0, "bundle")?)?;
371 
        let mut updates = Vec::new();
372 
        for u in o::array(field(m, 1, "updates")?, "updates")? {
373 
            let um = map(u, "update")?;
374 
            updates.push(RefUpdate {
375 
                name: o::text(field(um, 0, "name")?, "name")?.to_owned(),
376 
                old_state: o::oref(field(um, 1, "old_state")?, "old_state")?,
377 
                new_state: o::oref(field(um, 2, "new_state")?, "new_state")?,
378 
                force: field(um, 3, "force")?.as_bool().ok_or_else(|| schema("force: expected bool"))?,
379 
                attestation: opt_attest_from(field(um, 4, "attestation")?)?,
380 
            });
381 
        }
382 
        Ok((bundle, updates))
383 
    }
384
385 
    pub fn encode_push_response(results: &[RefResult]) -> Vec<u8> {
386 
        cbor::encode(&Value::map([(
387 
            0,
388 
            Value::Array(
389 
                results
390 
                    .iter()
391 
                    .map(|r| Value::map([(0, Value::Text(r.name.clone())), (1, Value::Bool(r.ok)), (2, Value::Text(r.reason.clone()))]))
392 
                    .collect(),
393 
            ),
394 
        )]))
395 
    }
396 
    pub fn decode_push_response(bytes: &[u8]) -> Result<Vec<RefResult>> {
397 
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
398 
        let m = map(&v, "push response")?;
399 
        let mut out = Vec::new();
400 
        for r in o::array(field(m, 0, "results")?, "results")? {
401 
            let rm = map(r, "result")?;
402 
            out.push(RefResult {
403 
                name: o::text(field(rm, 0, "name")?, "name")?.to_owned(),
404 
                ok: field(rm, 1, "ok")?.as_bool().ok_or_else(|| schema("ok: expected bool"))?,
405 
                reason: o::text(field(rm, 2, "reason")?, "reason")?.to_owned(),
406 
            });
407 
        }
408 
        Ok(out)
409 
    }
410 
}
411
412 
#[cfg(test)]
413 
mod tests {
414 
    use super::*;
415 
    use crate::attest::AttestBody;
416 
    use crate::objects::Hlc;
417
418 
    #[test]
419 
    fn wire_round_trips() {
420 
        let c = crate::objects::tests::sample_change();
421 
        let sk = crate::identity::SecretKey { ed_seed: [1; 32], ml_seed: [2; 32] };
422 
        let id = sk.identity("a", "b", 0);
423 
        let key = id.hash();
424 
        let sig = sk.sign(crate::identity::Domain::Change, &key, &c.hash());
425 
        let seal = Attest::seal([3; 16], vec![c.hash()]);
426 
        let seal_sig = sk.sign(crate::identity::Domain::Attest, &key, &seal.hash());
427 
        let bundle = Bundle {
428 
            changes: vec![(c.clone(), vec![sig.clone()])],
429 
            attests: vec![SignedAttest { attest: seal.clone(), sigs: vec![seal_sig] }],
430 
            objects: vec![(ObjectType::Identity, id.encode()), (ObjectType::Chunk, b"xyz".to_vec())],
431 
        };
432 
        let d = wire::decode_bundle(&wire::encode_bundle(&bundle)).unwrap();
433 
        assert_eq!(d.changes, bundle.changes);
434 
        assert_eq!(d.attests, bundle.attests);
435 
        assert_eq!(d.objects, bundle.objects);
436
437 
        let att = Attest { repo: [3; 16], hlc: Hlc::default(), body: AttestBody::Ref { name: "main".into(), old_state: ObjectRef([1; 32]), new_state: ObjectRef([2; 32]), serial: 4 } };
438 
        let refs = vec![
439 
            RemoteRef { name: "main".into(), state: ObjectRef([2; 32]), heads: vec![c.hash()], attestation: Some(SignedAttest { attest: att.clone(), sigs: vec![sig.clone()] }), last_serial: Some(4) },
440 
            RemoteRef { name: "dev".into(), state: ObjectRef([5; 32]), heads: vec![], attestation: None, last_serial: None },
441 
        ];
442 
        let (id_back, back) = wire::decode_refs(&wire::encode_refs([3; 16], &refs)).unwrap();
443 
        assert_eq!(id_back, [3; 16]);
444 
        assert_eq!(back.len(), 2);
445 
        assert_eq!(back[0].attestation.as_ref().unwrap().attest, att);
446 
        assert_eq!(back[1].last_serial, None);
447
448 
        let (h, w) = wire::decode_negotiate(&wire::encode_negotiate(&[ObjectRef([1; 32])], &[ObjectRef([2; 32])])).unwrap();
449 
        assert_eq!((h, w), (vec![ObjectRef([1; 32])], vec![ObjectRef([2; 32])]));
450 
        assert_eq!(wire::decode_hashes(&wire::encode_hashes(&[c.hash()])).unwrap(), vec![c.hash()]);
451
452 
        let up = RefUpdate { name: "main".into(), old_state: ObjectRef([1; 32]), new_state: ObjectRef([2; 32]), force: true, attestation: Some(SignedAttest { attest: att, sigs: vec![sig] }) };
453 
        let (b2, ups) = wire::decode_push(&wire::encode_push(&bundle, std::slice::from_ref(&up))).unwrap();
454 
        assert_eq!(b2.changes.len(), 1);
455 
        assert_eq!(ups[0].name, "main");
456 
        assert!(ups[0].force);
457 
        assert_eq!(ups[0].attestation, up.attestation);
458 
        let res = vec![RefResult { name: "main".into(), ok: false, reason: "stale".into() }];
459 
        assert_eq!(wire::decode_push_response(&wire::encode_push_response(&res)).unwrap(), res);
460 
        assert_eq!(wire::decode_capabilities(&wire::encode_capabilities([7; 16])).unwrap(), [7; 16]);
461 
    }
462 
}