crates/oxy-core/src/protocol.rs at 11c958021e1a · andymccall/oxygen

# 11c958021e1a oxygen / crates / oxy-core / src / protocol.rs

20.6 KiB · Rust · last changed Make the TLS client configuration public, so tools calling hub APIs trust exactly what oxy trusts 1 month ago source blame history raw

//! Transport-agnostic protocol (spec §16): what a client asks of a remote and
//! what a server does with a push. `LocalRemote` serves another repository
//! directory in-process, `HttpRemote` speaks to an `oxy-server` over HTTP(S),
//! and the server side reuses `Repo::receive_push` and friends unchanged.
//!
//! Wire messages are deterministic CBOR (`wire` module); over HTTP each is a
//! request or response body with content type `application/oxy-cbor`.

use crate::attest::Attest;
use crate::cbor::{self, Value};
use crate::hash::{ChangeHash, ObjectRef, ObjectType};
use crate::identity::Signature;
use crate::objects::{self as o, Change, ObjectError};
use crate::repo::{Repo, RepoError, Result};
use std::path::{Path, PathBuf};
use std::sync::Arc;

pub const CONTENT_TYPE: &str = "application/oxy-cbor";
pub const PROTOCOL_VERSION: u64 = 1;

/// An attestation together with the signatures over its ObjectRef.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct SignedAttest {
    pub attest: Attest,
    pub sigs: Vec<Signature>,
}

/// One reference as a remote reports it.
#[derive(Clone, Debug)]
pub struct RemoteRef {
    pub name: String,
    pub state: ObjectRef,
    pub heads: Vec<ChangeHash>,
    pub attestation: Option<SignedAttest>,
}

/// Everything needed to add a set of changes to a repository.
#[derive(Clone, Debug, Default)]
pub struct Bundle {
    /// In dependency order, each with its individual signatures.
    pub changes: Vec<(Change, Vec<Signature>)>,
    /// Seals covering the changes.
    pub attests: Vec<SignedAttest>,
    /// Chunks, blobs, identities, and state objects referenced by the changes.
    pub objects: Vec<(ObjectType, Vec<u8>)>,
}

impl Bundle {
    pub fn is_empty(&self) -> bool {
        self.changes.is_empty() && self.attests.is_empty() && self.objects.is_empty()
    }
}

/// A requested bookmark move (spec §10.2, §16.3).
#[derive(Clone, Debug)]
pub struct RefUpdate {
    pub name: String,
    pub old_state: ObjectRef,
    pub new_state: ObjectRef,
    pub force: bool,
    pub attestation: Option<SignedAttest>,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RefResult {
    pub name: String,
    pub ok: bool,
    pub reason: String,
}

pub trait Remote {
    fn repo_id(&self) -> Result<[u8; 16]>;
    fn refs(&self) -> Result<Vec<RemoteRef>>;
    /// Changes in the closure of `want` but not of `have`, in dependency order.
    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>>;
    fn fetch(&self, changes: &[ChangeHash]) -> Result<Bundle>;
    fn push(&self, bundle: Bundle, updates: Vec<RefUpdate>) -> Result<Vec<RefResult>>;
}

// ---------------------------------------------------------------- local

/// Another repository on this machine, served in-process.
pub struct LocalRemote {
    pub repo: Repo,
}

impl LocalRemote {
    pub fn open(path: &Path) -> Result<LocalRemote> {
        Ok(LocalRemote { repo: Repo::open(path)? })
    }
}

impl Remote for LocalRemote {
    fn repo_id(&self) -> Result<[u8; 16]> {
        self.repo.id()
    }
    fn refs(&self) -> Result<Vec<RemoteRef>> {
        self.repo.remote_refs()
    }
    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>> {
        self.repo.negotiate(have, want)
    }
    fn fetch(&self, changes: &[ChangeHash]) -> Result<Bundle> {
        self.repo.bundle_for(changes)
    }
    fn push(&self, bundle: Bundle, updates: Vec<RefUpdate>) -> Result<Vec<RefResult>> {
        self.repo.receive_push(bundle, updates)
    }
}

// ---------------------------------------------------------------- http

/// How to reach a remote: credentials and trust.
#[derive(Clone, Debug, Default)]
pub struct RemoteOptions {
    /// Bearer token for pushes.
    pub token: Option<String>,
    /// PEM file with extra root certificates to trust for `https://` remotes
    /// (self-hosted servers); the system web roots are always trusted.
    pub ca: Option<PathBuf>,
}

/// A remote served by `oxy-server` at `<base>/oxy/v1/...`.
pub struct HttpRemote {
    base: String,
    token: Option<String>,
    agent: ureq::Agent,
}

/// A TLS client configuration trusting the web roots plus `ca`, if given:
/// the trust every oxy connection uses, shared with tools that call hub APIs.
pub fn tls_client_config(ca: Option<&Path>) -> Result<Arc<rustls::ClientConfig>> {
    let mut roots = rustls::RootCertStore::empty();
    roots.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
    if let Some(path) = ca {
        let pem = std::fs::read(path).map_err(|e| RepoError::Msg(format!("reading CA file {}: {e}", path.display())))?;
        let certs = rustls_pemfile::certs(&mut pem.as_slice())
            .collect::<std::result::Result<Vec<_>, _>>()
            .map_err(|e| RepoError::Msg(format!("parsing CA file {}: {e}", path.display())))?;
        if certs.is_empty() {
            return Err(RepoError::Msg(format!("CA file {} holds no certificates", path.display())));
        }
        for c in certs {
            roots.add(c).map_err(|e| RepoError::Msg(format!("CA file {}: {e}", path.display())))?;
        }
    }
    let cfg = rustls::ClientConfig::builder_with_provider(Arc::new(rustls::crypto::ring::default_provider()))
        .with_safe_default_protocol_versions()
        .map_err(|e| RepoError::Msg(e.to_string()))?
        .with_root_certificates(roots)
        .with_no_client_auth();
    Ok(Arc::new(cfg))
}

impl HttpRemote {
    pub fn new(base: &str, options: &RemoteOptions) -> Result<HttpRemote> {
        let agent = ureq::AgentBuilder::new().tls_config(tls_client_config(options.ca.as_deref())?).build();
        Ok(HttpRemote { base: base.trim_end_matches('/').to_string(), token: options.token.clone(), agent })
    }

    fn call(&self, post: bool, endpoint: &str, body: Option<&[u8]>) -> Result<Vec<u8>> {
        let url = format!("{}/oxy/v1/{endpoint}", self.base);
        let mut req = if post { self.agent.post(&url) } else { self.agent.get(&url) };
        req = req.set("Accept", CONTENT_TYPE);
        if let Some(t) = &self.token {
            req = req.set("Authorization", &format!("Bearer {t}"));
        }
        let resp = match body {
            Some(b) => req.set("Content-Type", CONTENT_TYPE).send_bytes(b),
            None => req.call(),
        };
        match resp {
            Ok(r) => {
                let mut out = Vec::new();
                r.into_reader().read_to_end(&mut out).map_err(|e| RepoError::Msg(format!("{url}: {e}")))?;
                Ok(out)
            }
            Err(ureq::Error::Status(code, r)) => {
                let text = r.into_string().unwrap_or_default();
                Err(RepoError::Msg(format!("{url}: HTTP {code}: {}", text.trim())))
            }
            Err(e) => Err(RepoError::Msg(format!("{url}: {e}"))),
        }
    }
}

impl Remote for HttpRemote {
    fn repo_id(&self) -> Result<[u8; 16]> {
        wire::decode_capabilities(&self.call(false, "capabilities", None)?)
    }
    fn refs(&self) -> Result<Vec<RemoteRef>> {
        Ok(wire::decode_refs(&self.call(false, "refs", None)?)?.1)
    }
    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>> {
        wire::decode_hashes(&self.call(true, "negotiate", Some(&wire::encode_negotiate(have, want)))?)
    }
    fn fetch(&self, changes: &[ChangeHash]) -> Result<Bundle> {
        wire::decode_bundle(&self.call(true, "fetch", Some(&wire::encode_hashes(changes)))?)
    }
    fn push(&self, bundle: Bundle, updates: Vec<RefUpdate>) -> Result<Vec<RefResult>> {
        wire::decode_push_response(&self.call(true, "push", Some(&wire::encode_push(&bundle, &updates)))?)
    }
}

/// Open a remote by URL: a local path (optionally `file://`), or `http(s)://`.
pub fn open_remote(url: &str, options: &RemoteOptions) -> Result<Box<dyn Remote>> {
    if url.starts_with("http://") || url.starts_with("https://") {
        return Ok(Box::new(HttpRemote::new(url, options)?));
    }
    let path = url.strip_prefix("file://").unwrap_or(url);
    if path.contains("://") {
        return Err(RepoError::Msg(format!("unsupported remote URL {url:?}")));
    }
    Ok(Box::new(LocalRemote::open(Path::new(path))?))
}

// ---------------------------------------------------------------- wire

/// CBOR encodings of the protocol messages (spec §16.1).
pub mod wire {
    use super::*;

    fn schema(msg: &str) -> RepoError {
        RepoError::Object(ObjectError::Schema(msg.into()))
    }
    fn sigs_value(sigs: &[Signature]) -> Value {
        Value::Array(sigs.iter().map(|s| s.to_value()).collect())
    }
    fn sigs_from(v: &Value) -> Result<Vec<Signature>> {
        Ok(o::array(v, "signatures")?.iter().map(Signature::from_value).collect::<std::result::Result<Vec<_>, _>>()?)
    }
    fn hashes_value(hs: &[ObjectRef]) -> Value {
        Value::Array(hs.iter().map(|h| Value::Bytes(h.0.to_vec())).collect())
    }
    fn hashes_from(v: &Value) -> Result<Vec<ObjectRef>> {
        Ok(o::array(v, "hashes")?.iter().map(|h| o::oref(h, "hash")).collect::<std::result::Result<Vec<_>, _>>()?)
    }
    fn attest_value(sa: &SignedAttest) -> Value {
        Value::Array(vec![Value::Bytes(sa.attest.encode()), sigs_value(&sa.sigs)])
    }
    fn attest_from(v: &Value) -> Result<SignedAttest> {
        let a = o::array(v, "signed attest")?;
        if a.len() != 2 {
            return Err(schema("signed attest: expected [attest, sigs]"));
        }
        Ok(SignedAttest { attest: Attest::decode(o::bytes(&a[0], "attest")?)?, sigs: sigs_from(&a[1])? })
    }
    fn opt_attest_value(sa: &Option<SignedAttest>) -> Value {
        sa.as_ref().map(attest_value).unwrap_or(Value::Null)
    }
    fn opt_attest_from(v: &Value) -> Result<Option<SignedAttest>> {
        if v.is_null() { Ok(None) } else { attest_from(v).map(Some) }
    }
    fn map<'a>(v: &'a Value, what: &str) -> Result<&'a o::Map> {
        Ok(o::map(v, what)?)
    }
    fn field<'a>(m: &'a o::Map, k: u64, what: &str) -> Result<&'a Value> {
        Ok(o::field(m, k, what)?)
    }

    pub fn bundle_value(b: &Bundle) -> Value {
        Value::map([
            (
                0,
                Value::Array(b.changes.iter().map(|(c, s)| Value::Array(vec![Value::Bytes(c.encode()), sigs_value(s)])).collect()),
            ),
            (1, Value::Array(b.attests.iter().map(attest_value).collect())),
            (
                2,
                Value::Array(
                    b.objects.iter().map(|(t, body)| Value::Array(vec![Value::Text(t.name().into()), Value::Bytes(body.clone())])).collect(),
                ),
            ),
        ])
    }

    pub fn bundle_from_value(v: &Value) -> Result<Bundle> {
        let m = map(v, "bundle")?;
        let mut b = Bundle::default();
        for c in o::array(field(m, 0, "changes")?, "changes")? {
            let pair = o::array(c, "change entry")?;
            if pair.len() != 2 {
                return Err(schema("change entry: expected [change, sigs]"));
            }
            b.changes.push((Change::decode(o::bytes(&pair[0], "change")?)?, sigs_from(&pair[1])?));
        }
        for a in o::array(field(m, 1, "attests")?, "attests")? {
            b.attests.push(attest_from(a)?);
        }
        for ob in o::array(field(m, 2, "objects")?, "objects")? {
            let pair = o::array(ob, "object")?;
            if pair.len() != 2 {
                return Err(schema("object: expected [type, body]"));
            }
            let ty = ObjectType::from_name(o::text(&pair[0], "object type")?).ok_or_else(|| schema("unknown object type"))?;
            b.objects.push((ty, o::bytes(&pair[1], "object body")?.to_vec()));
        }
        Ok(b)
    }

    pub fn encode_bundle(b: &Bundle) -> Vec<u8> {
        cbor::encode(&bundle_value(b))
    }
    pub fn decode_bundle(bytes: &[u8]) -> Result<Bundle> {
        bundle_from_value(&cbor::decode(bytes).map_err(ObjectError::from)?)
    }

    pub fn encode_capabilities(id: [u8; 16]) -> Vec<u8> {
        cbor::encode(&Value::map([
            (0, Value::Array(vec![Value::Uint(PROTOCOL_VERSION)])),
            (1, Value::Bytes(id.to_vec())),
            (2, Value::Array(vec![Value::Text("seal".into())])),
        ]))
    }
    pub fn decode_capabilities(bytes: &[u8]) -> Result<[u8; 16]> {
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
        let m = map(&v, "capabilities")?;
        let versions = o::array(field(m, 0, "versions")?, "versions")?;
        if !versions.iter().any(|x| x.as_uint() == Some(PROTOCOL_VERSION)) {
            return Err(RepoError::Msg("remote does not speak protocol version 1".into()));
        }
        o::bytes(field(m, 1, "repo id")?, "repo id")?.try_into().map_err(|_| schema("repo id: expected 16 bytes"))
    }

    pub fn encode_refs(id: [u8; 16], refs: &[RemoteRef]) -> Vec<u8> {
        cbor::encode(&Value::map([
            (
                0,
                Value::Array(
                    refs.iter()
                        .map(|r| {
                            Value::map([
                                (0, Value::Text(r.name.clone())),
                                (1, Value::Bytes(r.state.0.to_vec())),
                                (2, hashes_value(&r.heads)),
                                (3, opt_attest_value(&r.attestation)),
                                (4, r.last_serial.map(Value::Uint).unwrap_or(Value::Null)),
                            ])
                        })
                        .collect(),
                ),
            ),
            (1, Value::Bytes(id.to_vec())),
        ]))
    }
    pub fn decode_refs(bytes: &[u8]) -> Result<([u8; 16], Vec<RemoteRef>)> {
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
        let m = map(&v, "refs")?;
        let id: [u8; 16] = o::bytes(field(m, 1, "repo id")?, "repo id")?.try_into().map_err(|_| schema("repo id: expected 16 bytes"))?;
        let mut refs = Vec::new();
        for r in o::array(field(m, 0, "refs")?, "refs")? {
            let rm = map(r, "ref")?;
            let ls = field(rm, 4, "last serial")?;
            refs.push(RemoteRef {
                name: o::text(field(rm, 0, "name")?, "name")?.to_owned(),
                state: o::oref(field(rm, 1, "state")?, "state")?,
                heads: hashes_from(field(rm, 2, "heads")?)?,
                attestation: opt_attest_from(field(rm, 3, "attestation")?)?,
                last_serial: if ls.is_null() { None } else { Some(o::uint(ls, "last serial")?) },
            });
        }
        Ok((id, refs))
    }

    pub fn encode_negotiate(have: &[ChangeHash], want: &[ChangeHash]) -> Vec<u8> {
        cbor::encode(&Value::map([(0, hashes_value(have)), (1, hashes_value(want))]))
    }
    pub fn decode_negotiate(bytes: &[u8]) -> Result<(Vec<ChangeHash>, Vec<ChangeHash>)> {
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
        let m = map(&v, "negotiate")?;
        Ok((hashes_from(field(m, 0, "have")?)?, hashes_from(field(m, 1, "want")?)?))
    }

    pub fn encode_hashes(hs: &[ChangeHash]) -> Vec<u8> {
        cbor::encode(&Value::map([(0, hashes_value(hs))]))
    }
    pub fn decode_hashes(bytes: &[u8]) -> Result<Vec<ChangeHash>> {
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
        hashes_from(field(map(&v, "hashes")?, 0, "hashes")?)
    }

    pub fn encode_push(bundle: &Bundle, updates: &[RefUpdate]) -> Vec<u8> {
        cbor::encode(&Value::map([
            (0, bundle_value(bundle)),
            (
                1,
                Value::Array(
                    updates
                        .iter()
                        .map(|u| {
                            Value::map([
                                (0, Value::Text(u.name.clone())),
                                (1, Value::Bytes(u.old_state.0.to_vec())),
                                (2, Value::Bytes(u.new_state.0.to_vec())),
                                (3, Value::Bool(u.force)),
                                (4, opt_attest_value(&u.attestation)),
                            ])
                        })
                        .collect(),
                ),
            ),
        ]))
    }
    pub fn decode_push(bytes: &[u8]) -> Result<(Bundle, Vec<RefUpdate>)> {
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
        let m = map(&v, "push")?;
        let bundle = bundle_from_value(field(m, 0, "bundle")?)?;
        let mut updates = Vec::new();
        for u in o::array(field(m, 1, "updates")?, "updates")? {
            let um = map(u, "update")?;
            updates.push(RefUpdate {
                name: o::text(field(um, 0, "name")?, "name")?.to_owned(),
                old_state: o::oref(field(um, 1, "old_state")?, "old_state")?,
                new_state: o::oref(field(um, 2, "new_state")?, "new_state")?,
                force: field(um, 3, "force")?.as_bool().ok_or_else(|| schema("force: expected bool"))?,
                attestation: opt_attest_from(field(um, 4, "attestation")?)?,
            });
        }
        Ok((bundle, updates))
    }

    pub fn encode_push_response(results: &[RefResult]) -> Vec<u8> {
        cbor::encode(&Value::map([(
            0,
            Value::Array(
                results
                    .iter()
                    .map(|r| Value::map([(0, Value::Text(r.name.clone())), (1, Value::Bool(r.ok)), (2, Value::Text(r.reason.clone()))]))
                    .collect(),
            ),
        )]))
    }
    pub fn decode_push_response(bytes: &[u8]) -> Result<Vec<RefResult>> {
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
        let m = map(&v, "push response")?;
        let mut out = Vec::new();
        for r in o::array(field(m, 0, "results")?, "results")? {
            let rm = map(r, "result")?;
            out.push(RefResult {
                name: o::text(field(rm, 0, "name")?, "name")?.to_owned(),
                ok: field(rm, 1, "ok")?.as_bool().ok_or_else(|| schema("ok: expected bool"))?,
                reason: o::text(field(rm, 2, "reason")?, "reason")?.to_owned(),
            });
        }
        Ok(out)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::attest::AttestBody;
    use crate::objects::Hlc;

    #[test]
    fn wire_round_trips() {
        let c = crate::objects::tests::sample_change();
        let sk = crate::identity::SecretKey { ed_seed: [1; 32], ml_seed: [2; 32] };
        let id = sk.identity("a", "b", 0);
        let key = id.hash();
        let sig = sk.sign(crate::identity::Domain::Change, &key, &c.hash());
        let seal = Attest::seal([3; 16], vec![c.hash()]);
        let seal_sig = sk.sign(crate::identity::Domain::Attest, &key, &seal.hash());
        let bundle = Bundle {
            changes: vec![(c.clone(), vec![sig.clone()])],
            attests: vec![SignedAttest { attest: seal.clone(), sigs: vec![seal_sig] }],
            objects: vec![(ObjectType::Identity, id.encode()), (ObjectType::Chunk, b"xyz".to_vec())],
        };
        let d = wire::decode_bundle(&wire::encode_bundle(&bundle)).unwrap();
        assert_eq!(d.changes, bundle.changes);
        assert_eq!(d.attests, bundle.attests);
        assert_eq!(d.objects, bundle.objects);

        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 } };
        let refs = vec![
            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) },
            RemoteRef { name: "dev".into(), state: ObjectRef([5; 32]), heads: vec![], attestation: None, last_serial: None },
        ];
        let (id_back, back) = wire::decode_refs(&wire::encode_refs([3; 16], &refs)).unwrap();
        assert_eq!(id_back, [3; 16]);
        assert_eq!(back.len(), 2);
        assert_eq!(back[0].attestation.as_ref().unwrap().attest, att);
        assert_eq!(back[1].last_serial, None);

        let (h, w) = wire::decode_negotiate(&wire::encode_negotiate(&[ObjectRef([1; 32])], &[ObjectRef([2; 32])])).unwrap();
        assert_eq!((h, w), (vec![ObjectRef([1; 32])], vec![ObjectRef([2; 32])]));
        assert_eq!(wire::decode_hashes(&wire::encode_hashes(&[c.hash()])).unwrap(), vec![c.hash()]);

        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] }) };
        let (b2, ups) = wire::decode_push(&wire::encode_push(&bundle, std::slice::from_ref(&up))).unwrap();
        assert_eq!(b2.changes.len(), 1);
        assert_eq!(ups[0].name, "main");
        assert!(ups[0].force);
        assert_eq!(ups[0].attestation, up.attestation);
        let res = vec![RefResult { name: "main".into(), ok: false, reason: "stale".into() }];
        assert_eq!(wire::decode_push_response(&wire::encode_push_response(&res)).unwrap(), res);
        assert_eq!(wire::decode_capabilities(&wire::encode_capabilities([7; 16])).unwrap(), [7; 16]);
    }
}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

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

Keyboard