//! 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; 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, } /// One reference as a remote reports it. #[derive(Clone, Debug)] pub struct RemoteRef { pub name: String, pub state: ObjectRef, pub heads: Vec, pub attestation: Option, } /// 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)>, /// Seals covering the changes. pub attests: Vec, /// Chunks, blobs, identities, and state objects referenced by the changes. pub objects: Vec<(ObjectType, Vec)>, } 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, } #[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>; /// Changes in the closure of `want` but not of `have`, in dependency order. fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result>; fn fetch(&self, changes: &[ChangeHash]) -> Result; fn push(&self, bundle: Bundle, updates: Vec) -> Result>; } // ---------------------------------------------------------------- local /// Another repository on this machine, served in-process. pub struct LocalRemote { pub repo: Repo, } impl LocalRemote { pub fn open(path: &Path) -> Result { Ok(LocalRemote { repo: Repo::open(path)? }) } } impl Remote for LocalRemote { fn repo_id(&self) -> Result<[u8; 16]> { self.repo.id() } fn refs(&self) -> Result> { self.repo.remote_refs() } fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result> { self.repo.negotiate(have, want) } fn fetch(&self, changes: &[ChangeHash]) -> Result { self.repo.bundle_for(changes) } fn push(&self, bundle: Bundle, updates: Vec) -> Result> { self.repo.receive_push(bundle, updates) } } // ---------------------------------------------------------------- http /// A remote served by `oxy-server` at `/oxy/v1/...`. pub struct HttpRemote { base: String, token: Option, agent: ureq::Agent, } impl HttpRemote { pub fn new(base: &str, token: Option) -> HttpRemote { HttpRemote { base: base.trim_end_matches('/').to_string(), token, agent: ureq::Agent::new() } } fn call(&self, post: bool, endpoint: &str, body: Option<&[u8]>) -> Result> { 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> { Ok(wire::decode_refs(&self.call(false, "refs", None)?)?.1) } fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result> { wire::decode_hashes(&self.call(true, "negotiate", Some(&wire::encode_negotiate(have, want)))?) } fn fetch(&self, changes: &[ChangeHash]) -> Result { wire::decode_bundle(&self.call(true, "fetch", Some(&wire::encode_hashes(changes)))?) } fn push(&self, bundle: Bundle, updates: Vec) -> Result> { 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, token: Option) -> Result> { if url.starts_with("http://") || url.starts_with("https://") { return Ok(Box::new(HttpRemote::new(url, token))); } 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> { Ok(o::array(v, "signatures")?.iter().map(Signature::from_value).collect::, _>>()?) } 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> { Ok(o::array(v, "hashes")?.iter().map(|h| o::oref(h, "hash")).collect::, _>>()?) } fn attest_value(sa: &SignedAttest) -> Value { Value::Array(vec![Value::Bytes(sa.attest.encode()), sigs_value(&sa.sigs)]) } fn attest_from(v: &Value) -> Result { 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) -> Value { sa.as_ref().map(attest_value).unwrap_or(Value::Null) } fn opt_attest_from(v: &Value) -> Result> { 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 { 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 { cbor::encode(&bundle_value(b)) } pub fn decode_bundle(bytes: &[u8]) -> Result { bundle_from_value(&cbor::decode(bytes).map_err(ObjectError::from)?) } pub fn encode_capabilities(id: [u8; 16]) -> Vec { 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 { 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)> { 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 { cbor::encode(&Value::map([(0, hashes_value(have)), (1, hashes_value(want))])) } pub fn decode_negotiate(bytes: &[u8]) -> Result<(Vec, Vec)> { 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 { cbor::encode(&Value::map([(0, hashes_value(hs))])) } pub fn decode_hashes(bytes: &[u8]) -> Result> { 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 { 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)> { 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 { 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> { 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]); } }