//! 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::policy::Policy; use crate::repo::{Repo, RepoError, Result}; use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex}; 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, } /// A policy together with the root signatures over its ObjectRef. #[derive(Clone, Debug, PartialEq, Eq)] pub struct SignedPolicy { pub policy: Policy, pub sigs: Vec, } impl SignedPolicy { /// The proposal-file encoding (`wire::policy_value`). pub fn encode(&self) -> Vec { crate::cbor::encode(&wire::policy_value(self)) } pub fn decode(b: &[u8]) -> Result { wire::policy_from( &crate::cbor::decode(b).map_err(|e| RepoError::Object(ObjectError::from(e)))?, ) } } /// What `refs` returns: every bookmark, every tag, and the current policy. #[derive(Clone, Debug, Default)] pub struct RefsResponse { pub refs: Vec, pub policy: Option, pub tags: Vec, /// Earlier policies, ascending by serial, so a client behind by more /// than one update can verify each step (spec §14.5). pub history: Vec, /// The bookmark a clone checks out (spec §16.1), when the remote names one. pub default_bookmark: Option, /// The host's latest transparency-log checkpoint, a signed note (spec /// §14.8), against which `RemoteRef::logged` proofs are made (key 6, /// only when the host keeps a log). pub log: Option, /// The witnesses the host asks to cosign its checkpoints: each one's /// name in cosignatures and its identity's bytes (key 7, only when it /// has any), which a client pins (spec §14.8, *Integrated time*). pub witnesses: Vec<(String, Vec)>, } /// Where a reference attestation sits in the host's transparency log /// (spec §14.8): its entry number, the entry, and the inclusion proof /// against the checkpoint served beside it. #[derive(Clone, Debug, PartialEq, Eq)] pub struct Logged { pub index: u64, pub entry: Vec, pub proof: Vec<[u8; 32]>, } /// A checkpoint witnesses cosigned that includes a logged entry (spec /// §14.8, *Integrated time*): the cosigned note, and the entry's inclusion /// proof against it. #[derive(Clone, Debug, PartialEq, Eq)] pub struct Integration { pub note: String, pub proof: Vec<[u8; 32]>, } /// A tag as a remote reports it (spec §10.1). #[derive(Clone, Debug, PartialEq, Eq)] pub struct TagRef { pub name: String, pub state: ObjectRef, pub heads: Vec, pub attestation: Option, } /// One reference as a remote reports it. #[derive(Clone, Debug)] pub struct RemoteRef { pub name: String, pub state: ObjectRef, pub heads: Vec, /// The latest attestation, only when it describes `state`. An unattested /// push (accepted while no trust policy exists) leaves it `None`. pub attestation: Option, /// Serial of the latest attestation ever recorded, stale or not, so the /// next attestation continues the sequence. pub last_serial: Option, /// The host's acceptance of `attestation` when a short-lived key signed /// it (spec §14.7): the time its certificate is checked at. pub accepted: Option, /// The attestation's place in the host's transparency log (key 6, only /// when it has one). pub logged: Option, /// The cosigned checkpoints that first included `logged`'s entry (key /// 7, only when there are any): what its integrated time is read from. pub integration: Vec, } /// 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)>, /// Policies between the receiver's and the one sent with the push, /// ascending by serial (spec §14.5). pub policies: Vec, /// Each change's hash, in the order of `changes`, when whoever made /// the bundle knows it: the store it came from, or the bytes it was /// decoded from. Never read from the wire. Empty when unknown. pub hashes: Vec, /// Changes as stored, with their signatures, in dependency order, for a /// fetch between repositories on one machine: their text blobs and /// chunks are among `objects`. Never on the wire. pub raw_changes: Vec<(Vec, Vec)>, /// Certificates of short-lived keys that signed what the bundle carries /// (spec §14.7), their issuers' identities among `objects`. pub certificates: 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>; /// The changes with their signatures, seals, identities, and blobs; blobs /// larger than `max_blob` bytes are left out for lazy fetching (spec §13.5). fn fetch(&self, changes: &[ChangeHash], max_blob: Option) -> Result; /// Individual objects by reference (spec §16.1 `objects`); a blob comes /// with its chunks. fn objects(&self, refs: &[ObjectRef]) -> Result)>>; /// A checkpoint of exactly `heads` (spec §16.1 `checkpoint`): the /// checkpoint object, its payload blob, and the blob's chunks. fn checkpoint(&self, heads: &[ChangeHash]) -> Result; /// The changes within `depth` dependency steps of `heads` (spec §17). fn shallow(&self, heads: &[ChangeHash], depth: u32) -> Result>; /// Push changes, bookmark updates, new tags, and optionally a newer /// policy. The policy is applied first; a result named `policy` reports /// it, and each tag reports as `tags/`. fn push( &self, bundle: Bundle, updates: Vec, policy: Option, tags: Vec, ) -> Result>; } // ---------------------------------------------------------------- local /// Another repository on this machine, served in-process. pub struct LocalRemote { pub repo: Repo, /// The pushing repository's id, which binds an unbound target on the /// first push, as `Oxy-Repo-Id` does over HTTP. pub claimed_id: Option<[u8; 16]>, } impl LocalRemote { pub fn open(path: &Path) -> Result { Ok(LocalRemote { repo: Repo::open(path)?, claimed_id: None, }) } } 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], max_blob: Option) -> Result { self.repo.bundle_raw_for(changes, max_blob) } fn objects(&self, refs: &[ObjectRef]) -> Result)>> { self.repo.objects_for(refs) } fn checkpoint(&self, heads: &[ChangeHash]) -> Result { self.repo.checkpoint_bundle_for(heads) } fn shallow(&self, heads: &[ChangeHash], depth: u32) -> Result> { self.repo.shallow_changes(heads, depth) } fn push( &self, bundle: Bundle, updates: Vec, policy: Option, tags: Vec, ) -> Result> { if self.repo.is_unbound() { match self.claimed_id { Some(id) => self.repo.bind_id(id)?, None => { return Err(RepoError::Msg( "the remote is unbound and this client did not say which repository it is" .into(), )) } } } self.repo.receive_push(bundle, updates, policy, tags) } } // ---------------------------------------------------------------- http /// How to reach a remote: credentials and trust. #[derive(Clone, Debug, Default)] pub struct RemoteOptions { /// Bearer token for pushes. pub token: Option, /// PEM file with extra root certificates to trust for `https://` remotes /// (self-hosted servers); the system web roots are always trusted. pub ca: Option, /// The local repository's id, sent as `Oxy-Repo-Id` so an unbound /// hosted repository can adopt it on the first push (spec §16.1). pub repo_id: Option<[u8; 16]>, /// Where a remote kept on a store (`dir://`, `s3://`, …) keeps its /// mirror and what it has seen: the remote's directory under /// `.oxy/remotes`. None (a clone, before there is a repository) means /// a temporary directory. pub cache_dir: Option, /// Other tokens for the same hub, tried in turn when `token` is /// refused (401, 403 or 404): those of enclosing checkouts, when /// `token` came from the credential store rather than from this /// remote's own configuration (oxygen#112). pub fallback_tokens: Vec, } /// Opens a store for a URL it understands, or passes (`None`). Stores that /// need more than the standard library (the cloud ones) are added by the /// program that links them, with [`register_store_opener`]. pub type StoreOpener = fn(&str, &RemoteOptions) -> Option>>; static STORE_OPENERS: Mutex> = Mutex::new(Vec::new()); pub fn register_store_opener(opener: StoreOpener) { STORE_OPENERS .lock() .unwrap_or_else(|p| p.into_inner()) .push(opener); } /// A store for `url`, from `dir://` or any registered opener. pub fn open_store( url: &str, options: &RemoteOptions, ) -> Option>> { if let Some(path) = url.strip_prefix("dir://") { return Some( crate::strata::LocalStore::open(path) .map(|s| Box::new(s) as Box), ); } let openers = STORE_OPENERS.lock().unwrap_or_else(|p| p.into_inner()); openers.iter().find_map(|open| open(url, options)) } /// The request header carrying the client's repository id. pub const REPO_ID_HEADER: &str = "Oxy-Repo-Id"; /// The id of a hosted repository that has not adopted one yet. pub const UNBOUND_ID: [u8; 16] = [0; 16]; /// A remote served by `oxy-server` at `/oxy/v1/...`. pub struct HttpRemote { base: String, /// The token in use, and those still to try when it is refused. token: Mutex>, fallbacks: Mutex>, repo_id: Option<[u8; 16]>, agent: ureq::Agent, } /// A TLS client configuration trusting the web roots plus `ca`, if given. fn tls_client_config(ca: Option<&Path>) -> Result> { 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::, _>>() .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 { refuse_plain_token(base, options)?; 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: Mutex::new(options.token.clone()), fallbacks: Mutex::new(options.fallback_tokens.iter().rev().cloned().collect()), repo_id: options.repo_id, agent, }) } /// A request, tried again with the next fallback token while the /// hub refuses the one in use (oxygen#112). fn call(&self, post: bool, endpoint: &str, body: Option<&[u8]>) -> Result> { loop { let token = self.token.lock().unwrap_or_else(|p| p.into_inner()).clone(); match self.call_once(post, endpoint, body, token.as_deref()) { Err(RepoError::Msg(m)) if token.is_some() && refused(&m) => { let next = self .fallbacks .lock() .unwrap_or_else(|p| p.into_inner()) .pop(); let Some(next) = next else { return Err(RepoError::Msg(m)); }; crate::credentials::warn_once(&format!( "{}: the stored token was refused; using the one an enclosing checkout has for this hub", self.base )); *self.token.lock().unwrap_or_else(|p| p.into_inner()) = Some(next); } other => return other, } } } fn call_once( &self, post: bool, endpoint: &str, body: Option<&[u8]>, token: Option<&str>, ) -> 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) = token { req = req.set("Authorization", &format!("Bearer {t}")); } if let Some(id) = &self.repo_id { req = req.set(REPO_ID_HEADER, &hex::encode(id)); } 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(); let msg = format!("{url}: HTTP {code}: {}", text.trim()); Err(RepoError::Msg(with_hint(msg))) } // The transport error already names the URL. Err(e) => Err(RepoError::Msg(with_hint(e.to_string()))), } } } /// Whether a failed request's message says the hub refused its token: /// HTTP 401, 403, or 404 (a hub hides a private repository from a token /// that may not read it). fn refused(message: &str) -> bool { [": HTTP 401", ": HTTP 403", ": HTTP 404"] .iter() .any(|c| message.contains(c)) } /// Refuse to send a token where anyone on the way could read it: plain /// `http` to another machine (see [`crate::credentials::token_allowed`]). fn refuse_plain_token(url: &str, options: &RemoteOptions) -> Result<()> { if options.token.is_some() && !crate::credentials::token_allowed(url) { return Err(RepoError::Msg(format!( "{url}: refusing to send a token over plain http; use https" ))); } Ok(()) } /// What to do about a failed request, recognised from its message. pub fn hint_for(message: &str) -> Option<&'static str> { let m = message.to_ascii_lowercase(); let hint = if m.contains("http 401") { "the hub needs you signed in, or did not accept the token: run `oxy auth login HUB` (pushing needs repo:write)" } else if m.contains("http 403") { "the token or account lacks permission: pushing needs a token with repo:write and write access to the repository" } else if m.contains("http 404") { "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`)" } else if m.contains("unknownissuer") || m.contains("unknown issuer") || m.contains("certificate") || m.contains("invalidcertificate") { "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..ca FILE`" } else if m.contains("connection refused") { "nothing is listening there: check the URL and port, and that the server is running" } else if m.contains("dns") || m.contains("failed to lookup") || m.contains("name or service not known") { "the host name does not resolve: check the URL" } else if m.contains("timed out") || m.contains("timeout") { "the server did not answer in time: check the network, or try again" } else { return None; }; Some(hint) } /// `message`, followed by what to do about it when that is known. pub fn with_hint(message: String) -> String { match hint_for(&message) { Some(h) => format!("{message}\n hint: {h}"), None => message, } } /// A JSON request to a hub API (`GET`, `POST` or `DELETE`), with the same /// TLS trust and token rules as remotes. Returns the status and body, an /// error status included; only transport failures are errors. pub fn http_json( method: &str, url: &str, body: Option<&str>, options: &RemoteOptions, ) -> Result<(u16, String)> { refuse_plain_token(url, options)?; let agent = ureq::AgentBuilder::new() .tls_config(tls_client_config(options.ca.as_deref())?) .timeout(std::time::Duration::from_secs(60)) .build(); let mut req = agent.request(method, url).set("Accept", "application/json"); if let Some(t) = &options.token { req = req.set("Authorization", &format!("Bearer {t}")); } let resp = match body { Some(b) => req.set("Content-Type", "application/json").send_string(b), None => req.call(), }; match resp { Ok(r) => { let code = r.status(); Ok((code, r.into_string().unwrap_or_default())) } Err(ureq::Error::Status(code, r)) => Ok((code, r.into_string().unwrap_or_default())), Err(e) => Err(RepoError::Msg(with_hint(e.to_string()))), } } /// An authenticated GET of any URL on a host, with the same TLS trust and /// token as remotes use (for hub APIs: proposals). Returns the body. pub fn http_get(url: &str, options: &RemoteOptions) -> Result> { refuse_plain_token(url, options)?; let agent = ureq::AgentBuilder::new() .tls_config(tls_client_config(options.ca.as_deref())?) .build(); let mut req = agent.get(url).set("Accept", "application/json"); if let Some(t) = &options.token { req = req.set("Authorization", &format!("Bearer {t}")); } match req.call() { 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(with_hint(format!( "{url}: HTTP {code}: {}", text.trim() )))) } // The transport error already names the URL. Err(e) => Err(RepoError::Msg(with_hint(e.to_string()))), } } 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], max_blob: Option) -> Result { wire::decode_bundle(&self.call( true, "fetch", Some(&wire::encode_fetch(changes, max_blob)), )?) } fn objects(&self, refs: &[ObjectRef]) -> Result)>> { Ok( wire::decode_bundle(&self.call(true, "objects", Some(&wire::encode_hashes(refs)))?)? .objects, ) } fn checkpoint(&self, heads: &[ChangeHash]) -> Result { wire::decode_bundle(&self.call(true, "checkpoint", Some(&wire::encode_hashes(heads)))?) } fn shallow(&self, heads: &[ChangeHash], depth: u32) -> Result> { wire::decode_hashes(&self.call( true, "shallow", Some(&wire::encode_shallow(heads, depth)), )?) } fn push( &self, bundle: Bundle, updates: Vec, policy: Option, tags: Vec, ) -> Result> { wire::decode_push_response(&self.call( true, "push", Some(&wire::encode_push( &bundle, &updates, policy.as_ref(), &tags, )), )?) } } /// A remote's URL as it is stored: an `http(s)://` URL as given, a local /// path made absolute (with its `file://` kept), so it still names the same /// repository from inside the clone (oxygen#57). pub fn stored_url(url: &str) -> String { // Only a path is made absolute: a plain one, or one behind file:// or // dir://. Every other URL (a hub, a bucket) is kept as written. let (prefix, path) = match (url.strip_prefix("file://"), url.strip_prefix("dir://")) { (Some(p), _) => ("file://", p), (_, Some(p)) => ("dir://", p), _ if url.contains("://") => return url.to_string(), _ => ("", url), }; match std::path::absolute(path) { Ok(abs) => format!("{prefix}{}", abs.display()), Err(_) => url.to_string(), } } /// Open a remote by URL: a local path (optionally `file://`), `http(s)://` /// (a hub), or a store (`dir://`, and whatever openers are registered). pub fn open_remote(url: &str, options: &RemoteOptions) -> Result> { if url.starts_with("http://") || url.starts_with("https://") { return Ok(Box::new(HttpRemote::new(url, options)?)); } if let Some(store) = open_store(url, options) { return crate::repo::strata::open_store_remote(store?, options); } let path = url.strip_prefix("file://").unwrap_or(url); if path.contains("://") { return Err(RepoError::Msg(format!("unsupported remote URL {url:?}"))); } let mut local = LocalRemote::open(Path::new(path))?; local.claimed_id = options.repo_id; Ok(Box::new(local)) } // ---------------------------------------------------------------- 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) } } /// `[policy bytes, [signature]]`: also the format of a policy proposal /// file exchanged between root holders (spec §14.5). pub fn policy_value(sp: &SignedPolicy) -> Value { Value::Array(vec![Value::Bytes(sp.policy.encode()), sigs_value(&sp.sigs)]) } pub fn policy_from(v: &Value) -> Result { let a = o::array(v, "signed policy")?; if a.len() != 2 { return Err(schema("signed policy: expected [policy, sigs]")); } Ok(SignedPolicy { policy: Policy::decode(o::bytes(&a[0], "policy")?)?, sigs: sigs_from(&a[1])?, }) } fn opt_policy_value(sp: Option<&SignedPolicy>) -> Value { sp.map(policy_value).unwrap_or(Value::Null) } fn opt_policy_from(v: &Value) -> Result> { if v.is_null() { Ok(None) } else { policy_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)?) } /// A bundle's value. Each change's text blobs and their chunks (spec /// §6.1) are added to its objects when they are not already there, so a /// bundle can always be read on its own. pub fn bundle_value(b: &Bundle) -> Value { let mut changes = Vec::with_capacity(b.raw_changes.len() + b.changes.len()); // Changes as stored are already their encoding, their texts among // the objects. for (bytes, s) in &b.raw_changes { changes.push(Value::Array(vec![ Value::Bytes(bytes.clone()), sigs_value(s), ])); } let mut texts: Vec<(ObjectType, Vec)> = Vec::new(); let mut listed: Option> = None; for (c, s) in &b.changes { let (bytes, stored) = c.encode_parts(); for t in stored { let listed = listed.get_or_insert_with(|| { b.objects .iter() .map(|(ty, body)| crate::hash::hash_object(*ty, body)) .collect() }); if !listed.insert(t.blob.hash()) { continue; } for chunk in crate::chunk::chunks(&t.content) { if listed.insert(crate::hash::hash_object(ObjectType::Chunk, chunk)) { texts.push((ObjectType::Chunk, chunk.to_vec())); } } texts.push((ObjectType::Blob, t.blob.encode())); } changes.push(Value::Array(vec![Value::Bytes(bytes), sigs_value(s)])); } let mut v = Value::map([ (0, Value::Array(changes)), ( 1, Value::Array(b.attests.iter().map(attest_value).collect()), ), ( 2, Value::Array( b.objects .iter() .chain(&texts) .map(|(t, body)| { Value::Array(vec![ Value::Text(t.name().into()), Value::Bytes(body.clone()), ]) }) .collect(), ), ), ( 3, Value::Array(b.policies.iter().map(policy_value).collect()), ), ]); // Only when there are any; older peers ignore the field. if !b.certificates.is_empty() { if let Value::Map(m) = &mut v { m.insert( 4, Value::Array( b.certificates .iter() .map(|c| Value::Bytes(c.encode())) .collect(), ), ); } } v } pub fn bundle_from_value(v: &Value) -> Result { let m = map(v, "bundle")?; let mut b = Bundle::default(); 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())); } // Stored inserts are read from the bundle's own objects. let mut by_ref: Option> = None; 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]")); } let encoded = o::bytes(&pair[0], "change")?; b.hashes .push(crate::hash::hash_object(ObjectType::Change, encoded)); let change = Change::decode_with(encoded, &mut |text, count, form| { let by_ref = by_ref.get_or_insert_with(|| { b.objects .iter() .enumerate() .map(|(i, (ty, body))| (crate::hash::hash_object(*ty, body), i)) .collect() }); let get = |r: &ObjectRef, want: ObjectType| match by_ref.get(r).map(|i| &b.objects[*i]) { Some((ty, body)) if *ty == want => Ok(body), _ => Err(ObjectError::Schema(format!( "the bundle lacks {} {}", want.name(), r.short(12) ))), }; let blob = crate::objects::Blob::decode(get(text, ObjectType::Blob)?)?; let mut content = Vec::with_capacity(blob.length as usize); for chunk in &blob.chunks { content.extend_from_slice(get(chunk, ObjectType::Chunk)?); } crate::objects::text_lines(&content, count, form) })?; b.changes.push((change, sigs_from(&pair[1])?)); } for a in o::array(field(m, 1, "attests")?, "attests")? { b.attests.push(attest_from(a)?); } if let Some(ps) = m.get(&3) { for p in o::array(ps, "policies")? { b.policies.push(policy_from(p)?); } } if let Some(cs) = m.get(&4) { for c in o::array(cs, "certificates")? { b.certificates.push( crate::cert::SignedCertificate::decode(o::bytes(c, "certificate")?) .map_err(|e| schema(&format!("certificate: {e}")))?, ); } } 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")) } fn tags_value(tags: &[TagRef]) -> Value { Value::Array( tags.iter() .map(|t| { Value::map([ (0, Value::Text(t.name.clone())), (1, Value::Bytes(t.state.0.to_vec())), (2, hashes_value(&t.heads)), (3, opt_attest_value(&t.attestation)), ]) }) .collect(), ) } fn tags_from(v: &Value) -> Result> { let mut out = Vec::new(); for t in o::array(v, "tags")? { let tm = map(t, "tag")?; out.push(TagRef { name: o::text(field(tm, 0, "name")?, "name")?.to_owned(), state: o::oref(field(tm, 1, "state")?, "state")?, heads: hashes_from(field(tm, 2, "heads")?)?, attestation: opt_attest_from(field(tm, 3, "attestation")?)?, }); } Ok(out) } pub fn encode_refs(id: [u8; 16], resp: &RefsResponse) -> Vec { let mut v = Value::map([ ( 0, Value::Array( resp.refs .iter() .map(|r| { let mut v = 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)), ]); // Only when there is one; older peers ignore it. if let Value::Map(m) = &mut v { if r.accepted.is_some() { m.insert(5, opt_attest_value(&r.accepted)); } if let Some(l) = &r.logged { m.insert( 6, Value::Array(vec![ Value::Uint(l.index), Value::Bytes(l.entry.clone()), proof_value(&l.proof), ]), ); } if !r.integration.is_empty() { m.insert( 7, Value::Array( r.integration .iter() .map(|i| { Value::Array(vec![ Value::Text(i.note.clone()), proof_value(&i.proof), ]) }) .collect(), ), ); } } v }) .collect(), ), ), (1, Value::Bytes(id.to_vec())), (2, opt_policy_value(resp.policy.as_ref())), (3, tags_value(&resp.tags)), ( 4, Value::Array(resp.history.iter().map(policy_value).collect()), ), ( 5, resp.default_bookmark .clone() .map(Value::Text) .unwrap_or(Value::Null), ), ]); if let Value::Map(m) = &mut v { if let Some(log) = &resp.log { m.insert(6, Value::Text(log.clone())); } if !resp.witnesses.is_empty() { m.insert( 7, Value::Array( resp.witnesses .iter() .map(|(name, id)| { Value::Array(vec![ Value::Text(name.clone()), Value::Bytes(id.clone()), ]) }) .collect(), ), ); } } cbor::encode(&v) } pub fn decode_refs(bytes: &[u8]) -> Result<([u8; 16], RefsResponse)> { 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")?) }, accepted: match rm.get(&5) { Some(a) => opt_attest_from(a)?, None => None, }, logged: match rm.get(&6) { Some(l) => { let a = o::array(l, "logged")?; if a.len() != 3 { return Err(schema("logged: expected [index, entry, proof]")); } Some(Logged { index: o::uint(&a[0], "index")?, entry: o::bytes(&a[1], "entry")?.to_vec(), proof: proof_from(&a[2])?, }) } None => None, }, integration: match rm.get(&7) { Some(l) => { let mut out = Vec::new(); for i in o::array(l, "integration")? { let a = o::array(i, "integration")?; if a.len() != 2 { return Err(schema("integration: expected [note, proof]")); } out.push(Integration { note: o::text(&a[0], "note")?.to_owned(), proof: proof_from(&a[1])?, }); } out } None => Vec::new(), }, }); } let policy = match m.get(&2) { Some(p) => opt_policy_from(p)?, None => None, }; let tags = match m.get(&3) { Some(t) => tags_from(t)?, None => Vec::new(), }; let mut history = Vec::new(); if let Some(h) = m.get(&4) { for p in o::array(h, "history")? { history.push(policy_from(p)?); } } let default_bookmark = match m.get(&5) { Some(Value::Text(t)) => Some(t.clone()), _ => None, }; let log = match m.get(&6) { Some(Value::Text(t)) => Some(t.clone()), _ => None, }; let mut witnesses = Vec::new(); if let Some(w) = m.get(&7) { for x in o::array(w, "witnesses")? { let a = o::array(x, "witness")?; if a.len() != 2 { return Err(schema("witness: expected [name, identity]")); } witnesses.push(( o::text(&a[0], "witness name")?.to_owned(), o::bytes(&a[1], "witness identity")?.to_vec(), )); } } Ok(( id, RefsResponse { refs, policy, tags, history, default_bookmark, log, witnesses, }, )) } /// A proof's hashes, from the leaf upwards (spec §14.8). fn proof_value(proof: &[[u8; 32]]) -> Value { Value::Array(proof.iter().map(|h| Value::Bytes(h.to_vec())).collect()) } fn proof_from(v: &Value) -> Result> { let mut proof = Vec::new(); for h in o::array(v, "proof")? { proof.push( o::bytes(h, "proof hash")? .try_into() .map_err(|_| schema("proof hash: expected 32 bytes"))?, ); } Ok(proof) } 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")?)?, )) } /// `fetch` request: `{0: [changes], 1: max blob size or null}`. pub fn encode_fetch(hs: &[ChangeHash], max_blob: Option) -> Vec { cbor::encode(&Value::map([ (0, hashes_value(hs)), (1, max_blob.map(Value::Uint).unwrap_or(Value::Null)), ])) } pub fn decode_fetch(bytes: &[u8]) -> Result<(Vec, Option)> { let v = cbor::decode(bytes).map_err(ObjectError::from)?; let m = map(&v, "fetch")?; let max = match m.get(&1) { Some(x) if !x.is_null() => Some(o::uint(x, "max blob")?), _ => None, }; Ok((hashes_from(field(m, 0, "changes")?)?, max)) } /// `shallow` request: `{0: [heads], 1: depth}`. pub fn encode_shallow(hs: &[ChangeHash], depth: u32) -> Vec { cbor::encode(&Value::map([ (0, hashes_value(hs)), (1, Value::Uint(depth as u64)), ])) } pub fn decode_shallow(bytes: &[u8]) -> Result<(Vec, u32)> { let v = cbor::decode(bytes).map_err(ObjectError::from)?; let m = map(&v, "shallow")?; Ok(( hashes_from(field(m, 0, "heads")?)?, o::u32_(field(m, 1, "depth")?, "depth")?, )) } 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], policy: Option<&SignedPolicy>, tags: &[TagRef], ) -> Vec { cbor::encode(&Value::map([ (0, bundle_value(bundle)), (2, opt_policy_value(policy)), (3, tags_value(tags)), ( 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(), ), ), ])) } /// A decoded push: bundle, bookmark updates, policy, tags. pub type PushParts = (Bundle, Vec, Option, Vec); pub fn decode_push(bytes: &[u8]) -> Result { 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")?)?, }); } let policy = match m.get(&2) { Some(p) => opt_policy_from(p)?, None => None, }; let tags = match m.get(&3) { Some(t) => tags_from(t)?, None => Vec::new(), }; Ok((bundle, updates, policy, tags)) } 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 policy = Policy::single_root([3; 16], id.clone(), 99); let psig = sk.sign(crate::identity::Domain::Policy, &key, &policy.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()), ], policies: vec![SignedPolicy { policy: policy.clone(), sigs: vec![psig.clone()], }], hashes: vec![], raw_changes: vec![], certificates: vec![], }; let d = wire::decode_bundle(&wire::encode_bundle(&bundle)).unwrap(); assert_eq!( d.hashes, vec![c.hash()], "the receiver hashes what it was sent" ); assert_eq!(d.changes, bundle.changes); assert_eq!(d.attests, bundle.attests); assert_eq!(d.objects, bundle.objects); assert_eq!(d.policies, bundle.policies); assert!(d.certificates.is_empty()); // Certificates (keyless, oxygenhub#241) travel as field 4, only // when there are any: a bundle without them encodes as before. let plain = wire::encode_bundle(&bundle); let mut with = bundle.clone(); let issuer_sk = crate::identity::SecretKey::generate(); let issuer = issuer_sk.identity("hub", "h@x", 1); with.certificates.push(crate::cert::SignedCertificate::sign( crate::cert::Certificate { key: id.clone(), issuer: issuer.hash(), subject: "agent:a@b".into(), claims: vec![], scope: crate::identity::Scope::CHANGES, not_before: 1, expires: 2, grant: None, }, &issuer, &issuer_sk, )); let encoded = wire::encode_bundle(&with); assert_ne!(encoded, plain); assert_eq!( wire::decode_bundle(&encoded).unwrap().certificates, with.certificates ); with.certificates.clear(); assert_eq!(wire::encode_bundle(&with), plain); 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), accepted: None, logged: None, integration: vec![], }, RemoteRef { name: "dev".into(), state: ObjectRef([5; 32]), heads: vec![], attestation: None, last_serial: None, accepted: None, logged: None, integration: vec![], }, ]; let tag = TagRef { name: "v1".into(), state: ObjectRef([2; 32]), heads: vec![c.hash()], attestation: None, }; let mut resp = RefsResponse { refs, policy: Some(SignedPolicy { policy: policy.clone(), sigs: vec![psig.clone()], }), tags: vec![tag.clone()], history: vec![SignedPolicy { policy: policy.clone(), sigs: vec![], }], default_bookmark: Some("develop".into()), log: Some("hub.example/log\n2\nroot\n\n- sig\n".into()), witnesses: vec![], }; // Without integrated times (spec §14.8) nothing is added: an old // peer's answer, and what it reads. let old = wire::encode_refs([3; 16], &resp); let back = wire::decode_refs(&old).unwrap().1; assert!(back.witnesses.is_empty() && back.refs.iter().all(|r| r.integration.is_empty())); resp.refs[1].logged = Some(Logged { index: 41, entry: vec![1, 2, 3], proof: vec![[9; 32], [8; 32]], }); resp.refs[1].integration = vec![Integration { note: "hub.example/log\n42\nroot\n\n- sig\n".into(), proof: vec![[7; 32]], }]; resp.witnesses = vec![("w.example".into(), vec![4, 5, 6])]; let new = wire::encode_refs([3; 16], &resp); let (id_back, back) = wire::decode_refs(&new).unwrap(); assert_eq!(back.log, resp.log); assert_eq!(back.refs[1].logged, resp.refs[1].logged); assert_eq!(back.refs[1].integration, resp.refs[1].integration); assert_eq!(back.witnesses, resp.witnesses); assert_eq!(back.refs[0].logged, None); // The new keys are 7s an old reader never asks for. let crate::cbor::Value::Map(m) = crate::cbor::decode(&new).unwrap() else { panic!() }; assert!(m.contains_key(&7)); assert_eq!(id_back, [3; 16]); assert_eq!(back.refs.len(), 2); assert_eq!(back.refs[0].attestation.as_ref().unwrap().attest, att); assert_eq!(back.refs[1].last_serial, None); assert_eq!(back.policy.as_ref().unwrap().policy, policy); assert_eq!(back.history, resp.history); assert_eq!(back.tags, vec![tag.clone()]); 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()] ); assert_eq!( wire::decode_fetch(&wire::encode_fetch(&[c.hash()], Some(4096))).unwrap(), (vec![c.hash()], Some(4096)) ); assert_eq!( wire::decode_fetch(&wire::encode_fetch(&[], None)).unwrap(), (vec![], None) ); assert_eq!( wire::decode_shallow(&wire::encode_shallow(&[c.hash()], 3)).unwrap(), (vec![c.hash()], 3) ); // An old-style fetch request without the filter field still decodes. assert_eq!( wire::decode_fetch(&wire::encode_hashes(&[c.hash()])).unwrap(), (vec![c.hash()], None) ); 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 sp = SignedPolicy { policy, sigs: vec![psig], }; let (b2, ups, pol, tags) = wire::decode_push(&wire::encode_push( &bundle, std::slice::from_ref(&up), Some(&sp), std::slice::from_ref(&tag), )) .unwrap(); assert_eq!(pol, Some(sp)); assert_eq!(tags, vec![tag]); 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] ); } }