//! Serving and talking to remotes (spec §10.2, §16). use super::tags::TagCheck; use super::*; /// What a push did, for hosts (hub spec §6.12). #[derive(Clone, Debug, Default)] pub struct PushReport { pub results: Vec, /// Accepted reference updates, in request order. pub updated: Vec, /// Changes this push stored that were not already present. pub imported: Vec, } #[derive(Clone, Debug)] pub struct RefUpdated { pub name: String, pub old_state: ObjectRef, pub new_state: ObjectRef, pub heads: Vec, } /// What a fetch brought in. #[derive(Debug, Clone, PartialEq, Eq, Default)] pub struct FetchReport { /// Bookmark name, its heads, and how many changes were new. pub refs: Vec<(String, Vec, usize)>, /// Tags adopted from the remote. pub new_tags: usize, /// Tags the remote withdrew, and so withdrawn here (oxygenhub#301). pub withdrawn_tags: Vec, /// The remote has never presented a policy, so nothing was verified /// against one (spec §14.5: warn loudly). pub unverified: bool, /// What checking the host's transparency log found short of a fork /// (spec §14.8): attestations not logged yet, a log that could not be /// read. A fork fails the fetch instead. pub log_notes: Vec, /// Bookmarks left as they were because their attestation could not be /// verified (its signer unknown here, or its signature bad), each with /// why (oxygenhub#266). The current bookmark is never skipped: its /// fetch fails instead. pub skipped: Vec<(String, String)>, } #[derive(Debug, Clone, PartialEq, Eq)] pub struct PullReport { /// The current bookmark's heads after the pull. pub heads: Vec, /// Whether the current bookmark changed. pub changed: bool, pub checked_out: bool, /// Other bookmarks the pull moved: each local bookmark tracks the /// remote's of the same name. pub others: Vec, /// What the fetch left as it was ([`FetchReport::skipped`]). pub skipped: Vec<(String, String)>, } /// The result a push of objects alone reports (a batch of a long history): /// not a reference. pub const OBJECTS_RESULT: &str = "objects"; /// Graph blobs asked for in one `objects` request when a shallow clone /// fetches its checkpoint in parts (#137): their chunk lists and chunks, /// a few MiB a request rather than one answer of gigabytes. const CHECKPOINT_PART: usize = 2048; impl Repo { // ------------------------------------------------------------ serving (spec §16) fn ref_attest_path(&self, bookmark: &str) -> PathBuf { self.oxy.join("attest").join(name_file(bookmark)) } /// The latest reference attestation stored for a bookmark. pub fn latest_ref_attest(&self, bookmark: &str) -> Result> { match fs::read_to_string(self.ref_attest_path(bookmark)) { Ok(h) => { let r = ObjectRef::from_hex(h.trim()) .ok_or_else(|| RepoError::Msg("corrupt attestation pointer".into()))?; Ok(Some(( r, SignedAttest { attest: self.store.get_attest(&r)?, sigs: self.store.signatures(&r)?, }, ))) } Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None), Err(e) => Err(e.into()), } } pub(crate) fn set_ref_attest(&self, bookmark: &str, r: &ObjectRef) -> Result<()> { fs::create_dir_all(self.oxy.join("attest"))?; write_atomic(&self.ref_attest_path(bookmark), r.to_hex().as_bytes()) } pub fn remote_refs(&self) -> Result { let view = self.view()?; let mut out = Vec::new(); for (name, state) in &view.bookmarks { let latest = self.latest_ref_attest(name)?.map(|(_, a)| a); let last_serial = latest.as_ref().and_then(|a| match a.attest.body { AttestBody::Ref { serial, .. } => Some(serial), _ => None, }); let attestation = latest.filter(|a| matches!(&a.attest.body, AttestBody::Ref { new_state, .. } if new_state == state)); let accepted = match &attestation { Some(a) => self.acceptance_of(&a.attest.hash())?, None => None, }; out.push(RemoteRef { name: name.clone(), state: *state, heads: self.store.get_state(state)?.heads, attestation, last_serial, accepted, logged: None, integration: vec![], }); } let policy = self.current_policy()?.map(|(_, sp)| sp); let history = self .policy_history()? .into_iter() .filter(|sp| { policy .as_ref() .is_some_and(|p| sp.policy.serial < p.policy.serial) }) .collect(); Ok(RefsResponse { refs: out, policy, tags: self.tag_refs()?, history, default_bookmark: Some(self.config.bookmark.clone()), log: None, witnesses: vec![], withdrawn: self.withdrawn_refs()?, }) } /// Every change reachable from `heads` that is in the store. fn closure_in_store(&self, heads: &[ChangeHash]) -> Result> { let mut seen = BTreeSet::new(); let mut stack: Vec = heads.to_vec(); while let Some(h) = stack.pop() { if seen.contains(&h) || !self.store.has(&h) { continue; } seen.insert(h); stack.extend(self.store.get_change_meta(&h)?.deps); } Ok(seen) } /// The changes within `depth` dependency steps of `heads` (spec §17), /// nearest first; depth 1 is the heads themselves. pub fn shallow_changes(&self, heads: &[ChangeHash], depth: u32) -> Result> { let mut out = Vec::new(); let mut seen: BTreeSet = BTreeSet::new(); let mut layer: Vec = heads.to_vec(); for _ in 0..depth.max(1) { let mut next = Vec::new(); for h in layer { if !seen.insert(h) || !self.store.has(&h) { continue; } let c = self.store.get_change(&h)?; next.extend(c.deps.iter().copied()); out.push(h); } if next.is_empty() { break; } layer = next; } out.reverse(); Ok(out) } fn shallow_path(&self) -> PathBuf { self.oxy.join("shallow") } /// The shallow boundary: dependencies this repository does not hold (spec §17). pub fn boundary(&self) -> BTreeSet { fs::read_to_string(self.shallow_path()) .map(|t| t.lines().filter_map(ObjectRef::from_hex).collect()) .unwrap_or_default() } pub fn is_shallow(&self) -> bool { !self.boundary().is_empty() } fn write_boundary(&self, boundary: &BTreeSet) -> Result<()> { if boundary.is_empty() { let _ = fs::remove_file(self.shallow_path()); return Ok(()); } let text: String = boundary .iter() .map(|h| format!("{}\n", h.to_hex())) .collect(); write_atomic(&self.shallow_path(), text.as_bytes()) } /// Recompute the boundary from what is stored: every dependency of a /// stored change that is not itself stored. fn refresh_boundary(&self, imported: &[ChangeHash]) -> Result<()> { let mut boundary = self.boundary(); boundary.retain(|h| !self.store.has(h)); for h in imported { for d in self.store.get_change_meta(h)?.deps { if !self.store.has(&d) { boundary.insert(d); } } } self.write_boundary(&boundary) } /// Fetch `depth` more levels of history below the shallow boundary. pub fn deepen(&self, remote_name: &str, depth: u32) -> Result { let boundary = self.boundary(); if boundary.is_empty() { return Ok(0); } let remote = self.open_named_remote(remote_name)?; let wanted = remote.shallow(&boundary.iter().copied().collect::>(), depth)?; let bundle = remote.fetch(&wanted, self.config.max_blob)?; let imported = self.import_bundle_shallow(&bundle, false)?; if self.config.max_blob.is_some() { self.register_missing_blobs(&imported, remote_name)?; } self.refresh_boundary(&imported)?; Ok(imported.len()) } /// `closure(want) \ closure(have)`, dependencies first (spec §16.2). pub fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result> { let have = self.closure_in_store(have)?; // In a shallow repository the walk stops at the boundary, whose // changes the remote is assumed to hold. let boundary = self.boundary(); let mut out = Vec::new(); let mut done: BTreeSet = have.clone(); done.extend(boundary.iter().copied()); // Iterative post-order DFS. let shallow = self.is_shallow(); let mut stack: Vec<(ChangeHash, bool)> = want.iter().rev().map(|h| (*h, false)).collect(); while let Some((h, expanded)) = stack.pop() { if done.contains(&h) { continue; } if !self.store.has(&h) { // A shallow repository holds only part of its history; a // change it lacks is below what it cloned (in a depth-0 // clone, any change of the checkpoint), which the remote // is assumed to hold, as it is for the boundary (#137). if shallow { done.insert(h); continue; } return fail(format!("change {} is not in this repository", h.short(12))); } if expanded { done.insert(h); out.push(h); continue; } stack.push((h, true)); for d in self.store.get_change_meta(&h)?.deps { if !done.contains(&d) { stack.push((d, false)); } } } Ok(out) } /// Bundle changes (in the given dependency order) with their signatures, /// seals, identities, blobs, chunks. pub fn bundle_for(&self, changes: &[ChangeHash]) -> Result { self.bundle_of(changes, None, false, false) } /// Objects by reference for the `objects` endpoint; a blob brings its chunks. pub fn objects_for(&self, refs: &[ObjectRef]) -> Result)>> { // In two reads in pack order (#137): a checkpoint's graph blobs come // this way in parts, tens of thousands for a big state. let mut found = Vec::with_capacity(refs.len()); for got in self.store.get_many(refs) { match got { Ok(x) => found.push(x), Err(StoreError::NotFound(_)) => {} Err(e) => return Err(e.into()), } } let mut chunks = Vec::new(); for (ty, body) in &found { if *ty == ObjectType::Blob { chunks.extend(Blob::decode(body)?.chunks); } } let mut out = Vec::with_capacity(found.len() + chunks.len()); for got in self.store.get_many(&chunks) { match got { Ok(x) => out.push(x), Err(StoreError::NotFound(_)) => {} Err(e) => return Err(e.into()), } } out.extend(found); Ok(out) } /// Like `bundle_for`, leaving out blobs larger than `max_blob` bytes /// (and their chunks) so the receiver fetches them lazily (spec §13.5). /// /// This is what a server answers a fetch with, so it also carries what /// verifies the attestations it serves beside it (see /// [`Repo::served_attest_signers`]), also when `changes` is empty. pub fn bundle_for_filtered( &self, changes: &[ChangeHash], max_blob: Option, ) -> Result { self.bundle_of(changes, max_blob, false, true) } /// Like [`Repo::bundle_for_filtered`] for a fetch between repositories /// on one machine: changes as they are stored ([`Bundle::raw_changes`]), /// with their text blobs and chunks among the objects, so neither side /// decodes or encodes their text. pub fn bundle_raw_for(&self, changes: &[ChangeHash], max_blob: Option) -> Result { self.bundle_of(changes, max_blob, true, true) } /// The keys that signed the attestations this repository serves with /// its references: each bookmark's latest and each tag's. A fetch /// bundle carries their identities and certificates (spec §16.2), so a /// client can check an attestation whose signer signed none of the /// changes it receives, a delegate or a short-lived key that only /// moved a bookmark (oxygenhub#266). One pointer and one signature /// list per reference; nothing per change. pub fn served_attest_signers(&self) -> Result> { let view = self.view()?; let pointers = view .bookmarks .keys() .map(|b| self.ref_attest_path(b)) .chain(view.tags.keys().map(|t| self.tag_attest_path(t))); let mut keys = BTreeSet::new(); for p in pointers { let Ok(text) = fs::read_to_string(&p) else { continue; }; let Some(r) = ObjectRef::from_hex(text.trim()) else { continue; }; keys.extend(self.store.signatures(&r)?.into_iter().map(|s| s.key)); } Ok(keys) } /// `serving`: a fetch's answer, which carries the identities and /// certificates of [`Repo::served_attest_signers`] too. fn bundle_of( &self, changes: &[ChangeHash], max_blob: Option, raw: bool, serving: bool, ) -> Result { let mut bundle = Bundle::default(); let mut objects: BTreeSet = BTreeSet::new(); let mut seals: BTreeSet = BTreeSet::new(); let add_object = |r: &ObjectRef, objects: &mut BTreeSet| { if !r.is_zero() && self.store.has(r) { objects.insert(*r); } }; for h in changes { // Raw: the stored bytes, and a skeleton for the rest. let (c, encoded) = if raw { let bytes = self.store.get_typed(h, ObjectType::Change)?; let (skel, texts) = Change::decode_skeleton(&bytes, *h)?; for t in texts { if let crate::objects::TextRef::Blob(r, _, _) = t { let blob = Blob::decode(&self.store.get_typed(&r, ObjectType::Blob)?)?; objects.insert(r); objects.extend(blob.chunks); } } (skel, Some(bytes)) } else { (self.store.get_change(h)?, None) }; let sigs = self.store.signatures(h)?; add_object(&c.author, &mut objects); for s in &sigs { add_object(&s.key, &mut objects); } // A replacement's new content only: its old content is the new // content of a change it depends on, which the receiver holds or // which this bundle carries (oxygen#136: removing ~1,100 binary // files sent them all again). for e in &c.edits { if let Edit::Replace { new: Some(b), .. } = e { if self.store.has(b) { if let Ok(blob) = Blob::decode(&self.store.get_typed(b, ObjectType::Blob)?) { if max_blob.is_some_and(|m| blob.length > m) { continue; } objects.insert(*b); objects.extend(blob.chunks.iter().copied()); } } } } for s in self.store.seals_for(h)? { seals.insert(s); } match encoded { Some(bytes) => bundle.raw_changes.push((bytes, sigs)), None => { bundle.changes.push((c, sigs)); bundle.hashes.push(*h); } } } for r in seals { let attest = self.store.get_attest(&r)?; let sigs = self.store.signatures(&r)?; for s in &sigs { add_object(&s.key, &mut objects); } bundle.attests.push(SignedAttest { attest, sigs }); } let signers: Vec = if serving { let signers = self.served_attest_signers()?; for k in &signers { add_object(k, &mut objects); } signers.into_iter().collect() } else { Vec::new() }; for r in objects { let (ty, body) = self.store.get(&r)?; bundle.objects.push((ty, body)); } self.attach_certificates_for(&mut bundle, &signers)?; Ok(bundle) } /// Add the certificates of short-lived keys that signed what `bundle` /// carries, and their issuers' identities, so a receiver can tell who /// they stand for (spec §14.7). `also` names more signers (a reference /// attestation sent beside the bundle). pub(crate) fn attach_certificates_for( &self, bundle: &mut Bundle, also: &[KeyId], ) -> Result<()> { let mut keys: BTreeSet = also.iter().copied().collect(); keys.extend( bundle .changes .iter() .flat_map(|(_, s)| s.iter().map(|s| s.key)), ); keys.extend( bundle .raw_changes .iter() .flat_map(|(_, s)| s.iter().map(|s| s.key)), ); keys.extend( bundle .attests .iter() .flat_map(|a| a.sigs.iter().map(|s| s.key)), ); for k in keys { let Some(r) = self.store.certificate_of(&k)? else { continue; }; let cert = self.store.get_certificate(&r)?; if bundle .certificates .iter() .any(|c| c.cert.hash() == cert.cert.hash()) { continue; } // An agent's certificate brings its grant's signer too, so the // receiver can check the grant (oxygen#134). let grantor = cert.cert.grant.as_ref().map(|g| g.grant.grantor); for id in [Some(k), Some(cert.cert.issuer), grantor].iter().flatten() { if let Ok(body) = self.store.get_typed(id, ObjectType::Identity) { if !bundle .objects .iter() .any(|(t, b)| *t == ObjectType::Identity && *b == body) { bundle.objects.push((ObjectType::Identity, body)); } } } bundle.certificates.push(cert); } Ok(()) } /// Keep the certificates a bundle brings once each verifies with the /// identity of the issuer it names; one that does not refuses the bundle. fn adopt_certificates(&self, certs: &[crate::cert::SignedCertificate]) -> Result<()> { for c in certs { let issuer = self.store.get_identity(&c.cert.issuer).map_err(|_| { RepoError::Msg(format!( "a certificate names issuer {}, whose identity did not come with it", c.cert.issuer.short(12) )) })?; c.verify(&issuer) .map_err(|e| RepoError::Msg(format!("a certificate does not verify: {e}")))?; self.store.put_identity(&c.cert.key)?; self.store.put_certificate(c)?; } Ok(()) } /// A signer authorized for `reference` by what an issuer certified it /// as (spec §14.7): a short-lived key with a kept certificate that /// verifies with its issuer, gives `scope`, covers `at_ms` (a time the /// signer did not choose) and names a subject the policy grants. pub(crate) fn certified_authorized( &self, p: &crate::policy::Policy, signer: &KeyId, reference: &str, scope: Scope, at_ms: u64, ) -> bool { let Ok(Some(r)) = self.store.certificate_of(signer) else { return false; }; let Ok(c) = self.store.get_certificate(&r) else { return false; }; let Some(issuer) = p .identity(&c.cert.issuer) .cloned() .or_else(|| self.store.get_identity(&c.cert.issuer).ok()) else { return false; }; c.verify(&issuer).is_ok() && c.cert.key_id() == *signer && c.cert.covers(scope, at_ms) && p.subject_grant(reference, &c.cert.issuer, &c.cert.subject, scope) } /// Store a bundle's contents. With `strict`, every change must /// authenticate (signature or seal by its author or a delegate). pub fn import_bundle(&self, bundle: &Bundle, strict: bool) -> Result> { self.import_bundle_inner(bundle, strict, false) } /// Like `import_bundle`, accepting changes whose dependencies are absent /// (they join the shallow boundary). pub fn import_bundle_shallow(&self, bundle: &Bundle, strict: bool) -> Result> { self.import_bundle_inner(bundle, strict, true) } fn import_bundle_inner( &self, bundle: &Bundle, strict: bool, shallow: bool, ) -> Result> { let batch = self.store.begin_batch()?; let result = self.import_bundle_batched(bundle, strict, shallow); batch.finish()?; result } fn import_bundle_batched( &self, bundle: &Bundle, strict: bool, shallow: bool, ) -> Result> { for (ty, _) in &bundle.objects { if !matches!( ty, ObjectType::Chunk | ObjectType::Blob | ObjectType::Identity | ObjectType::State | ObjectType::Checkpoint ) { return fail(format!("bundle may not carry {} objects", ty.name())); } } // All at once: in a large batch they are compressed on every core. let objects: Vec<(ObjectType, &[u8])> = bundle .objects .iter() .map(|(ty, body)| (*ty, body.as_slice())) .collect(); self.store.put_many(&objects)?; self.adopt_certificates(&bundle.certificates)?; for sa in &bundle.attests { let r = self.store.put_attest(&sa.attest)?; for sig in &sa.sigs { self.store.put_signature(&r, sig)?; } if let AttestBody::Seal { hashes } = &sa.attest.body { for h in hashes { self.store.index_seal(h, &r)?; } } } // Each change's hash: from the bundle when its maker knew it, else // worked out once (for a change adding a whole tree, hashing is // chunking all its text). let hashes: Vec = if bundle.hashes.len() == bundle.changes.len() { bundle.hashes.clone() } else { bundle.changes.iter().map(|(c, _)| c.hash()).collect() }; let raw_hashes: Vec = bundle .raw_changes .iter() .map(|(b, _)| hash_object(ObjectType::Change, b)) .collect(); let incoming: BTreeSet = hashes.iter().chain(&raw_hashes).copied().collect(); let boundary = self.boundary(); let mut imported = Vec::new(); // Changes as stored: checked, then stored as they are, their text // blobs having come with the objects. for ((bytes, sigs), h) in bundle.raw_changes.iter().zip(raw_hashes) { let (skel, texts) = Change::decode_skeleton(bytes, h)?; for d in &skel.deps { if !self.store.has(d) && !incoming.contains(d) && !shallow && !boundary.contains(d) { return fail(format!( "change {} depends on {}, which is neither stored nor in the bundle", h.short(12), d.short(12) )); } } for t in texts { if let crate::objects::TextRef::Blob(r, _, _) = t { if !self.store.has(&r) { return fail(format!( "change {} came without its text blob {}", h.short(12), r.short(12) )); } } } self.store.put(ObjectType::Change, bytes)?; for sig in sigs { self.store.put_signature(&h, sig)?; } imported.push(h); } for ((c, sigs), h) in bundle.changes.iter().zip(hashes) { for d in &c.deps { if !self.store.has(d) && !incoming.contains(d) && !shallow && !boundary.contains(d) { return fail(format!( "change {} depends on {}, which is neither stored nor in the bundle", h.short(12), d.short(12) )); } } let stored = self.store.put_change(c)?; if stored != h { return fail(format!( "change {} is not the change the bundle names", h.short(12) )); } for sig in sigs { self.store.put_signature(&h, sig)?; } imported.push(h); } if strict { for h in &imported { if !matches!(self.authenticate_change(h)?, SignatureStatus::Valid { .. }) { return fail(format!( "change {} is not authenticated by its author", h.short(12) )); } } } Ok(imported) } /// Verify a reference attestation against an update it claims to describe. /// `allow_equal` lets a client re-read the attestation it already has; a /// server receiving a push requires a strictly greater serial. /// A reference attestation signed by a delegate of an identity the /// rule lists is authorized as that identity (spec §14.5): the identity /// named the delegate with the `refs` scope, still in force at `at_ms` /// (the attestation's wall time), and the policy names the identity. /// Revoked keys authorize nothing either way. pub(crate) fn delegate_authorized( &self, p: &crate::policy::Policy, signer: &KeyId, reference: &str, at_ms: u64, ) -> bool { if p.is_revoked(signer) { return false; } let Some(rule) = p.rule_for(reference) else { return false; }; rule.keys.iter().any(|k| { !p.is_revoked(k) && p.identity(k) .cloned() .or_else(|| self.store.get_identity(k).ok()) .is_some_and(|id| id.delegates_to(signer, Scope::REFS, at_ms)) }) } /// A lasting attestation (a tag) signed by a key the policy has since /// rotated away from (spec §14.2) is authorized as the key that /// succeeded it: a key the rule lists reaches the signer through /// `previous` links, and the attestation is dated (`at_ms`) before the /// signer's successor was made, so a superseded key makes nothing new. /// Revoked keys authorize nothing. pub(crate) fn predecessor_authorized( &self, p: &crate::policy::Policy, signer: &KeyId, reference: &str, at_ms: u64, ) -> bool { if p.is_revoked(signer) { return false; } let Some(rule) = p.rule_for(reference) else { return false; }; let identity = |k: &KeyId| { p.identity(k) .cloned() .or_else(|| self.store.get_identity(k).ok()) }; rule.keys.iter().any(|k| { if p.is_revoked(k) { return false; } let mut cur = identity(k); for _ in 0..32 { let Some(id) = cur else { return false; }; match id.previous { Some(prev) if prev == *signer => return at_ms < id.created, Some(prev) if !p.is_revoked(&prev) => cur = identity(&prev), _ => return false, } } false }) } fn check_ref_attest( &self, sa: &SignedAttest, name: &str, old_state: &ObjectRef, new_state: &ObjectRef, last_serial: Option, allow_equal: bool, ) -> std::result::Result { let serial = self.check_ref_attest_body(sa, name, old_state, new_state, last_serial, allow_equal)?; self.verify_attest_signature(sa)?; Ok(serial) } /// What [`Repo::check_ref_attest`] checks short of the signature: the /// attestation describes this update of this repository with a serial /// past the last seen. fn check_ref_attest_body( &self, sa: &SignedAttest, name: &str, old_state: &ObjectRef, new_state: &ObjectRef, last_serial: Option, allow_equal: bool, ) -> std::result::Result { let AttestBody::Ref { name: n, old_state: o, new_state: nw, serial, } = &sa.attest.body else { return Err("attestation is not a ref attestation".into()); }; if n != name || o != old_state || nw != new_state { return Err("attestation does not describe this update".into()); } if sa.attest.repo != self.id().map_err(|e| e.to_string())? { return Err("attestation is for a different repository".into()); } if let Some(last) = last_serial { if *serial < last || (*serial == last && !allow_equal) { return Err(format!( "attestation serial {serial} is not greater than the last seen {last}" )); } } Ok(*serial) } /// An attestation's first signature verifies with its signer's /// identity, which must be stored here. fn verify_attest_signature(&self, sa: &SignedAttest) -> std::result::Result<(), String> { let r = sa.attest.hash(); let sig = sa.sigs.first().ok_or("attestation carries no signature")?; let id = self .store .get_identity(&sig.key) .map_err(|_| format!("unknown key {}", sig.key.short(12)))?; verify_signature(&id, sig, Domain::Attest, &r).map_err(|e| e.to_string()) } /// Accept a push (spec §16.4). Without a trust policy every repository /// is unverified, so unsigned changes and unattested updates are accepted; /// attestations that are present must still be valid. With a policy, /// every change and update is checked against it. A newer policy in the /// push is applied first when it is signed by the current roots. pub fn receive_push( &self, bundle: Bundle, updates: Vec, policy: Option, tags: Vec, ) -> Result> { Ok(self .receive_push_report(bundle, updates, policy, tags)? .results) } /// `receive_push` with what a host wants to log: each accepted /// reference's old and new state and heads, and the changes this push /// stored (hub spec §6.12). pub fn receive_push_report( &self, bundle: Bundle, updates: Vec, policy: Option, tags: Vec, ) -> Result { // Updates are checked against the head as it is now, not as this // handle first read it (oxygenhub#249): another push may have // landed since. self.refresh()?; let mut report = PushReport::default(); let results = &mut report.results; if let Some(sp) = policy { match self.adopt_policy_chain(&bundle.policies, &sp, "push received") { Ok(()) => { results.push(RefResult { name: "policy".into(), ok: true, reason: format!("serial {}", sp.policy.serial), }); } Err(e) => { results.push(RefResult { name: "policy".into(), ok: false, reason: e, }); for u in updates { results.push(RefResult { name: u.name, ok: false, reason: "policy update was rejected".into(), }); } return Ok(report); } } } let policy = self.current_policy()?.map(|(_, sp)| sp.policy); // A push of objects alone (one batch of a long history, oxygen#95) // names no reference: its outcome is reported as `objects`. let objects_only = updates.is_empty() && tags.is_empty(); match self.import_bundle(&bundle, policy.is_some()) { Ok(imported) => { if objects_only { results.push(RefResult { name: OBJECTS_RESULT.into(), ok: true, reason: format!("{} changes stored", imported.len()), }); } report.imported = imported; } Err(e) => { if objects_only { results.push(RefResult { name: OBJECTS_RESULT.into(), ok: false, reason: e.to_string(), }); } for u in updates { results.push(RefResult { name: u.name, ok: false, reason: e.to_string(), }); } return Ok(report); } } let (op_ref, op) = self.head_op()?; let mut view = op.view.clone(); let mut names = Vec::new(); for u in updates { let outcome = self.apply_update(&mut view, &u, policy.as_ref()); let (ok, reason) = match outcome { Ok(()) => (true, String::new()), Err(e) => (false, e), }; if ok { names.push(u.name.clone()); let heads = self .store .get_state(&u.new_state) .map(|s| s.heads) .unwrap_or_default(); report.updated.push(RefUpdated { name: u.name.clone(), old_state: u.old_state, new_state: u.new_state, heads, }); } results.push(RefResult { name: u.name, ok, reason, }); } for t in &tags { let name = format!("tags/{}", t.name); if t.is_withdrawal() { match self.adopt_withdrawal(&mut view, t, policy.as_ref(), &mut TagCheck::Received) { Ok(removed) => { // A withdrawal is an update to nothing, as a // bookmark's deletion is. if removed { report.updated.push(RefUpdated { name: name.clone(), old_state: t.state, new_state: ObjectRef::ZERO, heads: vec![], }); } names.push(name.clone()); results.push(RefResult { name, ok: true, reason: if removed { "withdrawn" } else { "already withdrawn" } .into(), }); } Err(e) => results.push(RefResult { name, ok: false, reason: e.reason, }), } continue; } let new = !view.tags.contains_key(&t.name); match self.adopt_tag(&mut view, t, policy.as_ref(), &mut TagCheck::Received) { Ok(()) => { // A tag new here is an update too, so a hub can start // its pipeline's tag jobs (oxygenflow#24); one it // already had, sent again, is not. if new { report.updated.push(RefUpdated { name: name.clone(), old_state: ObjectRef::ZERO, new_state: t.state, heads: t.heads.clone(), }); } names.push(name.clone()); results.push(RefResult { name, ok: true, reason: String::new(), }); } Err(e) => results.push(RefResult { name, ok: false, reason: e.reason, }), } } if !names.is_empty() { self.record_op( &[op_ref], &format!("push received: {}", names.join(" ")), view, )?; } Ok(report) } /// Check an update's attestation against the latest one for its /// reference and store it as the new latest. Failures say what was /// being done, for which reference, so a refused push can be diagnosed /// from its reason alone. fn record_update_attest( &self, u: &RefUpdate, sa: &SignedAttest, ) -> std::result::Result<(), String> { let name = &u.name; let last = self .latest_ref_attest(name) .map_err(|e| format!("reading the last attestation of {name:?}: {e}"))? .and_then(|(_, a)| match a.attest.body { AttestBody::Ref { serial, .. } => Some(serial), _ => None, }); self.check_ref_attest(sa, name, &u.old_state, &u.new_state, last, false)?; let store = |e: &dyn std::fmt::Display| format!("storing the attestation of {name:?}: {e}"); let r = self.store.put_attest(&sa.attest).map_err(|e| store(&e))?; for sig in &sa.sigs { self.store.put_signature(&r, sig).map_err(|e| store(&e))?; } self.set_ref_attest(name, &r).map_err(|e| store(&e))?; // Signed by a short-lived key: a host records when it accepted it. let certified = sa .sigs .first() .is_some_and(|s| self.store.certificate_of(&s.key).ok().flatten().is_some()); if certified { self.record_acceptance(&r).map_err(|e| store(&e))?; } Ok(()) } /// Record that this host accepted attestation `r` now, signed with its /// acceptor key (spec §14.7); nothing without one. pub(crate) fn record_acceptance(&self, r: &ObjectRef) -> Result<()> { let Some(acceptor) = &self.acceptor else { return Ok(()); }; let (id, sk) = (&acceptor.0, &acceptor.1); let attest = Attest { repo: self.id()?, hlc: Hlc::now(), body: AttestBody::Accepted { attest: *r }, }; let sig = sk.sign(Domain::Attest, &id.hash(), &attest.hash()); self.store.put_identity(id)?; let a = self.store.put_attest(&attest)?; self.store.put_signature(&a, &sig)?; self.store.index_add_pub("accepted", r, &a)?; Ok(()) } /// The host's acceptance of attestation `r`, if one was recorded. pub fn acceptance_of(&self, r: &ObjectRef) -> Result> { let Some(a) = self.store.index_get_pub("accepted", r)?.into_iter().next() else { return Ok(None); }; Ok(Some(SignedAttest { attest: self.store.get_attest(&a)?, sigs: self.store.signatures(&a)?, })) } /// When a certified signer's attestation `sa` is checked on a fetch: the /// host's acceptance time, if `accepted` is signed by the issuer of the /// signer's certificate and accepts `sa`; else the attestation's own /// time, which the signer chose. pub(crate) fn certified_time( &self, signer: &KeyId, sa: &SignedAttest, accepted: Option<&SignedAttest>, ) -> u64 { self.issuer_acceptance(signer, sa, accepted) .unwrap_or(sa.attest.hlc.wall_ms) } /// The time of `accepted` when it accepts exactly `sa` and is signed by /// the issuer of `signer`'s kept certificate (spec §14.7); else none. pub(crate) fn issuer_acceptance( &self, signer: &KeyId, sa: &SignedAttest, accepted: Option<&SignedAttest>, ) -> Option { let acc = accepted?; let c = self.store.certificate_of(signer).ok().flatten()?; let cert = self.store.get_certificate(&c).ok()?; let AttestBody::Accepted { attest } = &acc.attest.body else { return None; }; let sig = acc.sigs.iter().find(|s| s.key == cert.cert.issuer)?; let issuer = self.store.get_identity(&cert.cert.issuer).ok()?; (*attest == sa.attest.hash() && verify_signature(&issuer, sig, Domain::Attest, &acc.attest.hash()).is_ok()) .then_some(acc.attest.hlc.wall_ms) } /// Keep an acceptance this repository was given for attestation `r` /// (one [`Repo::issuer_acceptance`] took), so it is served again with /// what it accepts. pub(crate) fn keep_acceptance(&self, r: &ObjectRef, acc: &SignedAttest) -> Result<()> { let a = self.store.put_attest(&acc.attest)?; for sig in &acc.sigs { self.store.put_signature(&a, sig)?; } self.store.index_add_pub("accepted", r, &a)?; Ok(()) } /// A fetched bookmark's attestation by a signer the policy does not /// list itself: a delegate or a short-lived key, whose authority /// depends on when it signed (spec §14.5, §14.8). Checked at the /// attestation's integrated time when the host's log gives one; else /// as before, at the host's acceptance for a certified signer, when it /// countersigned one, or at the attestation's own time, and said so /// when the host keeps a log. Failing at the integrated time but not /// as before is a note while logs are new; a large gap between the /// claimed and the integrated time is a note. #[allow(clippy::too_many_arguments)] pub(crate) fn authorize_in_time( &self, p: &crate::policy::Policy, remote_name: &str, r: &crate::protocol::RemoteRef, sa: &SignedAttest, signer: &KeyId, log: &super::transparency::LogCheck, notes: &mut Vec, ) -> Result<()> { use super::transparency::{gap_note, judge, Timed}; let own = sa.attest.hlc.wall_ms; let at = self.certified_time(signer, sa, r.accepted.as_ref()); let time = log.time_of(&r.name); let check = |t_delegate: u64, t_cert: u64| { self.delegate_authorized(p, signer, &r.name, t_delegate) || self.certified_authorized(p, signer, &r.name, Scope::REFS, t_cert) }; let before = if at != own { "the host's acceptance" } else { "the time it claims" }; let what = format!("{remote_name}/{}", r.name); match judge(time.as_ref().ok().map(|i| i.ms()), |t| check(t, t), || check(own, at)) { Timed::Refused => { return fail(format!( "{what}: attested by {}, which the policy does not authorize for this bookmark; a policy owner can grant it with `oxy policy grant`", signer.short(12) )) } Timed::Integrated => {} Timed::Disagrees => notes.push(format!( "{what}: {}'s authority had lapsed when witness {} first saw its attestation, though not at {before}; accepted while transparency logs are new", signer.short(12), time.as_ref().map(|i| i.witness.as_str()).unwrap_or("?") )), Timed::Fallback if log.kept => notes.push(format!( "{what}: {} checked at {before}, not at a witnessed time: {}", signer.short(12), time.as_ref().err().map(String::as_str).unwrap_or("none") )), Timed::Fallback => {} } if let Some(n) = time.ok().and_then(|i| gap_note(own, i)) { notes.push(format!("{what}'s attestation {n}")); } Ok(()) } /// A deletion (spec §16.4): `new_state` is zero. The bookmark must exist /// at `old_state`, is never the bookmark a non-bare repository has /// checked out (a bare one's default is its host's to protect, see /// `oxy_server::Backend::default_bookmark`), and under a policy needs /// an attestation by a key that may update it. /// The deletion's attestation stays the bookmark's latest, so an older /// one cannot be replayed to bring it back. fn apply_deletion( &self, view: &mut View, u: &RefUpdate, policy: Option<&crate::policy::Policy>, ) -> std::result::Result<(), String> { let Some(current) = view.bookmarks.get(&u.name).copied() else { return Err(format!("no bookmark {:?} to delete", u.name)); }; if current != u.old_state { return Err("stale: the remote bookmark has moved; pull first".into()); } if !self.is_bare() && u.name == self.config.bookmark { return Err(format!( "{:?} is this repository's default bookmark (checked out there) and cannot be deleted", u.name )); } if let Some(p) = policy { let Some(sa) = &u.attestation else { return Err(format!( "deleting {:?} requires a signed reference attestation under this repository's policy", u.name )); }; let signer = sa .sigs .first() .map(|s| s.key) .ok_or("attestation carries no signature")?; // Received now: the receiver's own clock is the trusted time. let now = Hlc::now().wall_ms; if !p.authorizes(&signer, &u.name) && !self.delegate_authorized(p, &signer, &u.name, sa.attest.hlc.wall_ms) && !self.certified_authorized(p, &signer, &u.name, Scope::REFS, now) { return Err(format!( "key {} may not delete {:?} under this repository's policy", signer.short(12), u.name )); } } if let Some(sa) = &u.attestation { self.record_update_attest(u, sa)?; } view.bookmarks.remove(&u.name); Ok(()) } fn apply_update( &self, view: &mut View, u: &RefUpdate, policy: Option<&crate::policy::Policy>, ) -> std::result::Result<(), String> { if u.name == "policy" { return Err( "the policy reference is updated by sending a policy, not a bookmark update".into(), ); } let empty = self .store .put_state(&StateObject::new(vec![])) .map_err(|e| e.to_string())?; if u.new_state.is_zero() { return self.apply_deletion(view, u, policy); } let current = view.bookmarks.get(&u.name).copied().unwrap_or(empty); if current != u.old_state { return Err("stale: the remote bookmark has moved; pull first".into()); } let new_state = self .store .get_state(&u.new_state) .map_err(|_| "new state object is missing".to_string())?; for h in &new_state.heads { if !self.store.has(h) { return Err(format!("head {} is missing", h.short(12))); } } if !u.force { let old_heads = self .store .get_state(¤t) .map_err(|e| e.to_string())? .heads; let closure = self .closure_in_store(&new_state.heads) .map_err(|e| e.to_string())?; if let Some(h) = old_heads.iter().find(|h| !closure.contains(h)) { return Err(format!( "not a fast-forward: current head {} would be dropped (use --force)", h.short(12) )); } } if let Some(p) = policy { let Some(sa) = &u.attestation else { return Err(format!( "{:?} requires a signed reference attestation under this repository's policy", u.name )); }; let signer = sa .sigs .first() .map(|s| s.key) .ok_or("attestation carries no signature")?; // Received now: the receiver's own clock is the trusted time. let now = Hlc::now().wall_ms; if !p.authorizes(&signer, &u.name) && !self.delegate_authorized(p, &signer, &u.name, sa.attest.hlc.wall_ms) && !self.certified_authorized(p, &signer, &u.name, Scope::REFS, now) { return Err(format!( "key {} may not update {:?} under this repository's policy", signer.short(12), u.name )); } let old_heads = self .store .get_state(¤t) .map_err(|e| e.to_string())? .heads; self.state_allowed(&old_heads, &new_state.heads, p, &u.name)?; } if let Some(sa) = &u.attestation { self.record_update_attest(u, sa)?; } view.bookmarks.insert(u.name.clone(), u.new_state); Ok(()) } // ------------------------------------------------------------ client side fn remote_url(&self, name: &str) -> Result { self.config.remotes.get(name).cloned().ok_or_else(|| { let names: Vec<&str> = self.config.remotes.keys().map(String::as_str).collect(); RepoError::Msg(match names.as_slice() { [] => format!("no remote named {name:?}; add one with `oxy remote add`"), _ if name == "origin" && names.len() > 1 && self.config.default_remote.is_none() => format!( "this repository has no remote named origin, and several others ({}): name one (`oxy pull {}`), or make one the default with `oxy config core.remote {}`", names.join(", "), names[0], names[0] ), _ => format!( "no remote named {name:?}; this repository has {}", names.join(", ") ), }) }) } /// Refuse a remote that is another repository. An **unbound** remote /// (id all zero: a hosted repository nobody has pushed to) is accepted /// for pushes, which bind it to this repository's id (spec §16.1). fn check_remote_id( &self, remote: &dyn crate::protocol::Remote, remote_name: &str, allow_unbound: bool, ) -> Result<()> { let theirs = remote.repo_id()?; if theirs == self.id()? { return Ok(()); } if theirs == crate::protocol::UNBOUND_ID { if allow_unbound { return Ok(()); } return fail(format!( "remote {remote_name:?} is empty and unbound: push to it first" )); } fail(format!( "remote {remote_name:?} is a different repository (repository IDs differ); check its URL with `oxy remote list`" )) } /// Credentials and trust for a remote: `remote..token` and /// `remote..ca`, else the defaults for its URL /// ([`Repo::fill_credentials`]). pub(crate) fn remote_options(&self, name: &str) -> RemoteOptions { let url = self.config.remotes.get(name).cloned().unwrap_or_default(); let mut options = RemoteOptions { repo_id: self.id().ok(), token: self.config.remote_tokens.get(name).cloned(), ca: self.config.remote_cas.get(name).map(PathBuf::from), cache_dir: Some(self.oxy.join("remotes").join(name).join("strata")), ..RemoteOptions::default() }; self.fill_credentials(&url, &mut options); options } /// The user's credential store (`credentials.toml` beside the user /// configuration file). pub fn credentials(&self) -> crate::credentials::Credentials { crate::credentials::Credentials::at( self.user_config .parent() .map(Path::to_path_buf) .unwrap_or_else(default_home), ) } /// Fill in what `options` lacks for `url`: `$OXY_TOKEN` and `$OXY_CA`, /// then the user's credential store for its hub, then a remote on the /// same hub in an enclosing checkout (a subrepo borrows its parent's). /// When the token is not this remote's own or the environment's, the /// other tokens enclosing checkouts hold for the hub are kept to try /// if it is refused: a stale stored login must not hide a parent's /// good token (oxygen#112). fn fill_credentials(&self, url: &str, options: &mut RemoteOptions) { if options.token.is_some() && options.ca.is_some() { return; } let own = options.token.is_some(); let env = std::env::var("OXY_TOKEN").is_ok_and(|t| !t.is_empty()); let explicit = own || env; let stored = self.credentials().options_for(url); // A checkout's own token refused as expired or unknown (401) falls // back to this computer's sign-in for the hub, never to another // checkout's, and never when it is refused for permission (#155). if own && !env { if let Some(t) = stored .token .clone() .filter(|t| options.token.as_ref() != Some(t)) { options.fallback_tokens.push(t); options.fallback_on_401_only = true; } } options.token = options.token.take().or(stored.token); options.ca = options.ca.take().or(stored.ca); let enclosing = self.enclosing_credentials(url); if options.token.is_none() { options.token = enclosing.iter().find_map(|(t, _)| t.clone()); } if options.ca.is_none() { options.ca = enclosing.iter().find_map(|(_, c)| c.clone()); } if !explicit { for t in enclosing.into_iter().filter_map(|(t, _)| t) { if options.token.as_ref() != Some(&t) && !options.fallback_tokens.contains(&t) { options.fallback_tokens.push(t); } } } } /// What to do when `remote`'s own token was refused (`message` says /// HTTP 401) and so was this computer's sign-in for its hub, which it /// falls back to (oxygen#155). pub fn stale_token_note(&self, remote: &str, message: &str) -> Option { if !message.contains("HTTP 401") { return None; } let own = self.config.remote_tokens.get(remote)?; let url = self.config.remotes.get(remote)?; let stored = self.credentials().options_for(url).token?; (stored != *own).then(|| { format!( "the hub refused both this checkout's own token for {remote} (remote.{remote}.token) and your sign-in: \ `oxy auth login HUB`, then `oxy config --unset remote.{remote}.token` so the sign-in is used" ) }) } /// The tokens and CAs of remotes on `url`'s hub in enclosing /// checkouts, nearest first. fn enclosing_credentials(&self, url: &str) -> Vec<(Option, Option)> { let mut out = Vec::new(); let Some(hub) = crate::credentials::hub_of(url) else { return out; }; let mut at = self.root.parent(); while let Some(dir) = at { let Ok(cfg) = read_config_file(&dir.join(OXY_DIR).join("config")) else { break; }; for (key, remote_url) in &cfg { let Some(name) = key .strip_prefix("remote.") .and_then(|k| k.strip_suffix(".url")) else { continue; }; if crate::credentials::hub_of(remote_url).as_deref() != Some(hub.as_str()) { continue; } out.push(( cfg.get(&format!("remote.{name}.token")).cloned(), cfg.get(&format!("remote.{name}.ca")).map(PathBuf::from), )); } at = dir.parent(); } out } /// Push seal attestations on their own (hub spec §7.4): a bundle with /// no changes and no reference updates, which the receiver stores. pub fn push_seals(&self, remote_name: &str, seals: &[ObjectRef]) -> Result<()> { let remote = self.open_named_remote(remote_name)?; self.check_remote_id(remote.as_ref(), remote_name, false)?; let mut bundle = Bundle::default(); for r in seals { let attest = self.store.get_attest(r)?; let sigs = self.store.signatures(r)?; bundle .attests .push(crate::protocol::SignedAttest { attest, sigs }); } let results = remote.push(bundle, vec![], None, vec![])?; if let Some(r) = results.iter().find(|r| !r.ok) { return fail(format!("{}: {}", r.name, r.reason)); } Ok(()) } /// The remote whose URL starts with `base` (a hub), if configured. pub fn remote_for_url(&self, base: &str) -> Option { let base = base.trim_end_matches('/'); self.config .remotes .iter() .find(|(_, url)| url.trim_end_matches('/') == base) .map(|(name, _)| name.clone()) } /// Open a remote by name or, failing that, by URL with the default /// credentials (see [`Repo::credentials_for`]). pub fn open_remote_at(&self, url: &str) -> Result<(String, Box)> { if let Some(name) = self.remote_for_url(url) { return Ok((name.clone(), self.open_named_remote(&name)?)); } let options = RemoteOptions { repo_id: self.id().ok(), ..self.credentials_for(url) }; Ok((url.to_string(), open_remote(url, &options)?)) } /// Credentials for a URL: those of the configured remote with that /// URL, else the environment's. pub fn options_for_url(&self, url: &str) -> RemoteOptions { match self.remote_for_url(url) { Some(name) => self.remote_options(&name), None => RemoteOptions { repo_id: self.id().ok(), ..self.credentials_for(url) }, } } pub(crate) fn open_named_remote(&self, name: &str) -> Result> { open_remote(&self.remote_url(name)?, &self.remote_options(name)) } fn serial_path(&self, remote: &str, bookmark: &str) -> PathBuf { self.oxy .join("remotes") .join(remote) .join(format!("{bookmark}.serial")) } fn last_seen_serial(&self, remote: &str, bookmark: &str) -> Option { fs::read_to_string(self.serial_path(remote, bookmark)) .ok() .and_then(|s| s.trim().parse().ok()) } fn record_serial(&self, remote: &str, bookmark: &str, serial: u64) -> Result<()> { let p = self.serial_path(remote, bookmark); fs::create_dir_all(p.parent().unwrap())?; write_atomic(&p, serial.to_string().as_bytes()) } fn policy_serial_path(&self, remote: &str) -> PathBuf { self.oxy.join("remotes").join(remote).join("policy.serial") } /// The highest policy serial `remote` has ever presented. fn remote_policy_serial(&self, remote: &str) -> Option { fs::read_to_string(self.policy_serial_path(remote)) .ok() .and_then(|s| s.trim().parse().ok()) } fn record_policy_serial(&self, remote: &str, serial: u64) -> Result<()> { let p = self.policy_serial_path(remote); fs::create_dir_all(p.parent().unwrap())?; write_atomic(&p, serial.to_string().as_bytes()) } fn lazy_dir(&self) -> PathBuf { self.oxy.join("lazy") } /// Remember that `blob` was withheld by `remote` (spec §13.5). fn register_lazy(&self, blob: &ObjectRef, remote: &str) -> Result<()> { fs::create_dir_all(self.lazy_dir())?; write_atomic(&self.lazy_dir().join(blob.to_hex()), remote.as_bytes()) } /// Which remote promised an absent blob, if any. pub fn lazy_remote(&self, blob: &ObjectRef) -> Option { fs::read_to_string(self.lazy_dir().join(blob.to_hex())) .ok() .map(|s| s.trim().to_string()) } /// Fetch an absent blob (and its chunks) from the remote that promised it. pub(crate) fn fetch_lazy_blob(&self, blob: &ObjectRef) -> Result<()> { let Some(remote_name) = self.lazy_remote(blob) else { return Err(RepoError::Store(StoreError::NotFound(*blob))); }; let remote = self.open_named_remote(&remote_name)?; let objects = remote.objects(&[*blob])?; for (ty, body) in objects { if matches!(ty, ObjectType::Blob | ObjectType::Chunk) { self.store.put(ty, &body)?; } } if !self.store.has(blob) { return fail(format!( "{remote_name} did not send blob {}", blob.short(12) )); } let _ = fs::remove_file(self.lazy_dir().join(blob.to_hex())); Ok(()) } /// After importing from `remote`, record every blob the changes reference /// that the store lacks as lazily available from that remote. fn register_missing_blobs(&self, changes: &[ChangeHash], remote: &str) -> Result { let mut n = 0; for h in changes { let c = self.store.get_change(h)?; for e in &c.edits { if let Edit::Replace { old, new, .. } = e { for b in [old, new].into_iter().flatten() { if !self.store.has(b) { self.register_lazy(b, remote)?; n += 1; } } } } } Ok(n) } /// Every head this repository knows: bookmarks, remote tracking, working copy. pub fn known_heads(&self, view: &View) -> Result> { let mut have: BTreeSet = BTreeSet::new(); for st in view .bookmarks .values() .chain(view.remotes.values().flat_map(|b| b.values())) { if let Ok(s) = self.store.get_state(st) { have.extend(s.heads); } } have.extend(view.workspaces.values().copied()); Ok(have.into_iter().collect()) } /// Fetch every bookmark of a remote into `remotes//bookmarks/*` (spec §10.2). pub fn fetch(&self, remote_name: &str) -> Result { let remote = self.open_named_remote(remote_name)?; self.check_remote_id(remote.as_ref(), remote_name, false)?; let resp = { let _t = crate::timing::phase("fetch: refs"); remote.refs()? }; // The host's transparency log (spec §14.8): a fork fails here. let log_phase = crate::timing::phase("fetch: transparency log"); let log = self.check_log( &self.remote_url(remote_name)?, self.config.remote_cas.get(remote_name).map(Path::new), &resp, )?; drop(log_phase); let mut log_notes = log.notes.clone(); // Trust policy (spec §14.5). Adopt the remote's policy when it is newer // and signed by roots we trust, or on first contact. A remote that // once presented a policy may never present none or an older one: // that is a downgrade. A remote that never presented one is // unverified, and this fetch is checked only as far as possible. let local_policy = self.current_policy()?.map(|(_, sp)| sp.policy); let seen = self.remote_policy_serial(remote_name); let mut unverified = false; match &resp.policy { Some(sp) => { if seen.is_some_and(|s| sp.policy.serial < s) { return fail(format!("{remote_name} presents policy serial {} but presented {} before: refusing the downgrade", sp.policy.serial, seen.unwrap())); } match &local_policy { Some(l) if sp.policy.serial < l.serial => {} Some(l) if sp.policy.serial == l.serial => { if sp.policy.hash() != l.hash() { return fail(format!( "{remote_name} presents a different policy with the same serial {}", l.serial )); } } _ => { self.adopt_policy_chain( &resp.history, sp, &format!("oxy fetch {remote_name}"), ) .map_err(|e| RepoError::Msg(format!("policy from {remote_name}: {e}")))?; } } self.record_policy_serial(remote_name, sp.policy.serial)?; } None => { if seen.is_some() { return fail(format!("{remote_name} presented a policy before but has none now: refusing the downgrade")); } unverified = true; } } let policy = if unverified { None } else { self.current_policy()?.map(|(_, sp)| sp.policy) }; let (op_ref, op) = self.head_op()?; let mut view = op.view.clone(); let have = self.known_heads(&view)?; let mut report = Vec::new(); let mut skipped = Vec::new(); let max_blob = self.config.max_blob; // One negotiation and one bundle for every bookmark's heads at // once (oxygen#158): a bookmark at a time cost a round trip each and // sent again the changes bookmarks share, as `have` did not grow // between them (160 bookmarks: 160 bundles, 49k changes encoded for // a few hundred). Heads already here, with every attestation's // signer known, need no round trip at all. let mut wanted: Vec = resp .refs .iter() .flat_map(|r| r.heads.iter().copied()) .filter(|h| !self.store.has(h)) .collect(); wanted.sort(); wanted.dedup(); // A bundle also carries the identities and certificates of the keys // that signed the remote's attestations (spec §16.2): with no change // wanted, ask anyway when one of those signers is unknown here. let unknown_signer = resp .refs .iter() .filter_map(|r| r.attestation.as_ref()) .chain(resp.tags.iter().filter_map(|t| t.attestation.as_ref())) .flat_map(|a| a.sigs.iter().map(|s| s.key)) .any(|k| { self.store.get_identity(&k).is_err() && !matches!(self.store.certificate_of(&k), Ok(Some(_))) }); let shallow = self.is_shallow(); let imported: Vec = if wanted.is_empty() && !unknown_signer { Vec::new() } else { let needed = { let _t = crate::timing::phase("fetch: negotiate"); remote.negotiate(&have, &wanted)? }; let bundle = { let _t = crate::timing::phase("fetch: bundle"); remote.fetch(&needed, max_blob)? }; // Changes are not individually re-verified here: each bookmark's // attestation (checked below) binds its whole state, and the // receiver of the push verified the added changes at that time. // A shallow repository takes changes whose dependencies it never // held (below what it cloned), as its clone did (#137). let _t = crate::timing::phase("fetch: import"); if shallow { self.import_bundle_shallow(&bundle, false)? } else { self.import_bundle(&bundle, false)? } }; if shallow { self.refresh_boundary(&imported)?; } if max_blob.is_some() { self.register_missing_blobs(&imported, remote_name)?; } let boundary = if shallow { self.boundary() } else { BTreeSet::new() }; // Each bookmark's new changes: those just imported that its heads // reach, for the report. let mut new_deps: HashMap> = HashMap::new(); for h in &imported { if let Ok(m) = self.store.get_change_meta(h) { new_deps.insert(*h, m.deps); } } let new_in = |heads: &[ChangeHash]| -> usize { let mut seen: HashSet = HashSet::new(); let mut stack: Vec = heads .iter() .filter(|h| new_deps.contains_key(*h)) .copied() .collect(); while let Some(h) = stack.pop() { if seen.insert(h) { stack.extend( new_deps[&h] .iter() .filter(|d| new_deps.contains_key(*d) && !seen.contains(*d)), ); } } seen.len() }; let checks = crate::timing::phase("fetch: check each bookmark"); for r in resp.refs { for h in &r.heads { if !self.store.has(h) && !boundary.contains(h) { return fail(format!("remote did not send head {}", h.short(12))); } } let state = self.store.put_state(&StateObject::new(r.heads.clone()))?; if state != r.state { return fail(format!( "remote state {} does not match its heads", r.state.short(12) )); } if let Some(p) = &policy { let Some(sa) = &r.attestation else { return fail(format!( "{remote_name}/{} has no attestation, which its policy requires", r.name )); }; let signer = sa .sigs .first() .map(|s| s.key) .ok_or_else(|| RepoError::Msg("attestation carries no signature".into()))?; if !p.authorizes(&signer, &r.name) { self.authorize_in_time(p, remote_name, &r, sa, &signer, &log, &mut log_notes)?; } } if let Some(sa) = &r.attestation { let AttestBody::Ref { old_state, new_state, .. } = &sa.attest.body else { return fail("remote sent a non-ref attestation for a bookmark"); }; let (old_state, new_state) = (*old_state, *new_state); let serial = self .check_ref_attest_body( sa, &r.name, &old_state, &new_state, self.last_seen_serial(remote_name, &r.name), true, ) .map_err(|e| { RepoError::Msg(format!("attestation for {}/{}: {e}", remote_name, r.name)) })?; if new_state != r.state { return fail(format!( "attestation for {}/{} does not cover its current state", remote_name, r.name )); } // A signature this client cannot verify (its signer unknown // here, as from a host that did not send the identity, or // bad) fails that bookmark alone: it stays as it was and // the rest are fetched (oxygenhub#266). The current // bookmark, which a pull would take, still fails the fetch. // A policy's requirements were checked above and still fail. if let Err(e) = self.verify_attest_signature(sa) { let why = format!("attestation for {remote_name}/{}: {e}", r.name); if r.name == self.config.bookmark { return fail(why); } skipped.push((r.name, why)); continue; } let ar = self.store.put_attest(&sa.attest)?; for sig in &sa.sigs { self.store.put_signature(&ar, sig)?; } self.record_serial(remote_name, &r.name, serial)?; } view.remotes .entry(remote_name.to_string()) .or_default() .insert(r.name.clone(), state); let n = new_in(&r.heads); report.push((r.name, r.heads, n)); } drop(checks); let tags_phase = crate::timing::phase("fetch: tags"); let mut new_tags = 0; for t in &resp.tags { if view.tags.contains_key(&t.name) { continue; } // Withdrawn here: the name is not used again. if self.tag_withdrawal(&t.name)?.is_some() { continue; } // The tag's changes must be here: fetch them if the bookmarks did not bring them. let missing: Vec = t .heads .iter() .filter(|h| !self.store.has(h)) .copied() .collect(); if !missing.is_empty() { let needed = remote.negotiate(&self.known_heads(&view)?, &missing)?; let bundle = remote.fetch(&needed, max_blob)?; self.import_bundle(&bundle, false)?; } self.store.put_state(&StateObject::new(t.heads.clone()))?; let mut check = TagCheck::Fetched { remote: remote_name, log: &log, notes: &mut log_notes, }; match self.adopt_tag(&mut view, t, policy.as_ref(), &mut check) { Ok(()) => new_tags += 1, // Nothing says when a short-lived key signed it: the tag is // left out, said so, and the rest is fetched (oxygen#129). Err(e) if e.untimed => { skipped.push((format!("tags/{}", t.name), e.reason)); } Err(e) => return fail(format!("{remote_name} {}: {}", t.name, e.reason)), } } // The remote withdrew a tag this repository has: so does this one, // once the withdrawal is checked as a push's would be; one that is // not is left as it is, said so (oxygenhub#301). drop(tags_phase); let mut withdrawn_tags = Vec::new(); for t in &resp.withdrawn { if !view.tags.contains_key(&t.name) { continue; } let mut check = TagCheck::Fetched { remote: remote_name, log: &log, notes: &mut log_notes, }; match self.adopt_withdrawal(&mut view, t, policy.as_ref(), &mut check) { Ok(true) => withdrawn_tags.push(t.name.clone()), Ok(false) => {} Err(e) => skipped.push(( format!("tags/{}", t.name), format!("withdrawn on {remote_name}, kept here: {}", e.reason), )), } } if view != op.view { self.record_op(&[op_ref], &format!("oxy fetch {remote_name}"), view)?; } let _ = self.run_hook( "post-fetch", &format!( "{{\"event\":\"post-fetch\",\"remote\":{},\"bookmarks\":{}}}", json_str(remote_name), json_array(report.iter().map(|(n, heads, _)| format!( "{{\"name\":{},\"heads\":{}}}", json_str(n), json_array(heads.iter().map(|h| json_str(&h.to_hex()))) ))) ), ); Ok(FetchReport { refs: report, unverified, new_tags, withdrawn_tags, log_notes, skipped, }) } /// Shallow fetch of the remote's bookmark: its checkpoint as our base, /// the changes within `depth` steps of its heads, and the boundary below. pub fn fetch_shallow(&self, remote_name: &str, depth: u32) -> Result { let remote = self.open_named_remote(remote_name)?; self.check_remote_id(remote.as_ref(), remote_name, false)?; let resp = remote.refs()?; let bookmark = self.config.bookmark.clone(); if let Some(sp) = &resp.policy { self.adopt_policy_chain(&resp.history, sp, "oxy clone --depth") .map_err(|e| RepoError::Msg(format!("policy from {remote_name}: {e}")))?; self.record_policy_serial(remote_name, sp.policy.serial)?; } let Some(r) = resp.refs.into_iter().find(|r| r.name == bookmark) else { return fail(format!("{remote_name} has no bookmark {bookmark:?}")); }; if r.heads.is_empty() { return Ok(0); } // Base: the remote's checkpoint of exactly these heads, without its // graph blobs, which come in parts below (#137). let cp_bundle = { let _t = crate::timing::phase("shallow: checkpoint request"); remote.checkpoint_index(&r.heads)? }; // In one batch: a big state's checkpoint is tens of thousands of // graph blobs and chunks, which stored one by one were each written // loose and flushed (a VS Code clone ran at 27 objects a second). let objects: Vec<(ObjectType, &[u8])> = cp_bundle .objects .iter() .map(|(ty, body)| (*ty, body.as_slice())) .collect(); let t_store = crate::timing::phase("shallow: store the checkpoint"); let batch = self.store.begin_batch()?; let stored = self.store.put_many(&objects); batch.finish()?; drop(t_store); let cp_ref = objects .iter() .zip(stored?) .find(|((ty, _), _)| *ty == ObjectType::Checkpoint) .map(|(_, r)| r); let Some(cp_ref) = cp_ref else { return fail(format!("{remote_name} sent no checkpoint")); }; // The graph blobs the payload names, with their chunks, a part at // a time (#137): one answer for a big state was gigabytes, held // whole on both sides. A remote that sent the whole checkpoint has // left none to ask for. let cp = crate::objects::Checkpoint::decode( &self.store.get_typed(&cp_ref, ObjectType::Checkpoint)?, )?; let graphs: Vec = self .checkpoint_graph_blobs(&cp)? .into_iter() .filter(|b| !self.store.has(b)) .collect(); { let _t = crate::timing::phase("shallow: the checkpoint's graphs, in parts"); for part in graphs.chunks(CHECKPOINT_PART) { let objects = remote.objects(part)?; let objects: Vec<(ObjectType, &[u8])> = objects .iter() .filter(|(ty, _)| matches!(ty, ObjectType::Blob | ObjectType::Chunk)) .map(|(ty, body)| (*ty, body.as_slice())) .collect(); let batch = self.store.begin_batch()?; let stored = self.store.put_many(&objects); batch.finish()?; stored?; } if let Some(b) = graphs.iter().find(|b| !self.store.has(b)) { return fail(format!( "{remote_name} did not send graph blob {}", b.short(12) )); } } self.install_checkpoint(&bookmark, &cp_ref)?; let _t_blobs = crate::timing::phase("shallow: the binary files' blobs"); // The blobs of the state's opaque files come with no change when the // window holds none of the changes that set them (#137): fetch them // now, or under `filter.max-blob` leave them to be fetched on read. let base = State::from_checkpoint_history(&self.read_blob(&cp.payload)?)?; let mut missing: Vec = base .opaque_nodes() .iter() .flat_map(|n| base.live_blobs(n)) .filter(|b| !self.store.has(b)) .collect(); missing.sort(); missing.dedup(); if self.config.max_blob.is_some() { for b in &missing { self.register_lazy(b, remote_name)?; } } else { for part in missing.chunks(512) { let objects = remote.objects(part)?; let objects: Vec<(ObjectType, &[u8])> = objects .iter() .filter(|(ty, _)| matches!(ty, ObjectType::Blob | ObjectType::Chunk)) .map(|(ty, body)| (*ty, body.as_slice())) .collect(); let batch = self.store.begin_batch()?; let stored = self.store.put_many(&objects); batch.finish()?; stored?; } if let Some(b) = missing.iter().find(|b| !self.store.has(b)) { return fail(format!("{remote_name} did not send blob {}", b.short(12))); } } // History window. Depth 0 takes none (#137): the checkpoint holds // every change's metadata, and its heads become the boundary, so a // state of many heads (33,835 in the VS Code import) costs nothing // to fetch. let imported = if depth == 0 { let mut boundary = self.boundary(); boundary.extend(r.heads.iter().filter(|h| !self.store.has(h)).copied()); self.write_boundary(&boundary)?; Vec::new() } else { let wanted = remote.shallow(&r.heads, depth)?; let bundle = remote.fetch(&wanted, self.config.max_blob)?; self.import_bundle_shallow(&bundle, false)? }; if self.config.max_blob.is_some() { self.register_missing_blobs(&imported, remote_name)?; } self.refresh_boundary(&imported)?; if let Some(sa) = &r.attestation { let ar = self.store.put_attest(&sa.attest)?; for sig in &sa.sigs { self.store.put_signature(&ar, sig)?; } if let AttestBody::Ref { serial, .. } = sa.attest.body { self.record_serial(remote_name, &r.name, serial)?; } } let state = self.store.put_state(&StateObject::new(r.heads.clone()))?; let (op_ref, op) = self.head_op()?; let mut view = op.view.clone(); view.remotes .entry(remote_name.to_string()) .or_default() .insert(r.name.clone(), state); self.record_op( &[op_ref], &format!("oxy fetch {remote_name} --depth {depth}"), view, )?; Ok(imported.len()) } /// Fetch, then take the union of the local bookmark's state with the /// remote's: no merge change, conflicts become reified (spec §9.4). pub fn pull(&self, remote_name: &str) -> Result { let _wc = self.lock_working_copy()?; let skipped = self.fetch(remote_name)?.skipped; let (op_ref, op) = self.head_op()?; let current = self.config.bookmark.clone(); let mut local_current = self.heads_of(&op.view, ¤t)?; local_current.sort(); let mut view = op.view.clone(); let mut current_heads = local_current.clone(); let mut changed = false; let mut others = Vec::new(); let tracked = op .view .remotes .get(remote_name) .cloned() .unwrap_or_default(); for (name, remote_state) in tracked { // Every local bookmark tracks the remote's of the same name; // the current one may be new here. let local_state = op.view.bookmarks.get(&name).copied(); if local_state.is_none() && name != current { continue; } if local_state == Some(remote_state) { continue; } let local = self.heads_of(&op.view, &name)?; let remote_heads = self.store.get_state(&remote_state)?.heads; if remote_heads.iter().all(|h| local.contains(h)) { continue; } let mut all = local.clone(); all.extend(remote_heads); let st = self.load_state_history(&all)?; let mut heads = st.maximal(&all); heads.sort(); let mut sorted_local = local; sorted_local.sort(); if heads == sorted_local { continue; } view.bookmarks.insert( name.clone(), self.store.put_state(&StateObject::new(heads.clone()))?, ); if name == current { changed = true; current_heads = heads; } else { others.push(name); } } if changed || !others.is_empty() { self.record_op(&[op_ref], &format!("oxy pull {remote_name}"), view)?; } // The working copy follows its bookmark whenever nothing would be // lost, also when this pull brought nothing new but it had been // left behind (oxygen#89). let checked_out = self.follow_bookmark(¤t_heads, ¤t, "oxy pull")?; Ok(PullReport { heads: if changed { current_heads } else { local_current }, changed, checked_out, others, skipped, }) } /// How many changes the working copy's bookmark has that its base /// lacks, when the base is part of the bookmark's state (it fell behind, /// and `oxy checkout` or `oxy pull` brings it up); `None` when it is on /// the bookmark or holds changes the bookmark lacks. History is read /// only when the base and the bookmark's heads differ (oxygen#89). pub fn working_copy_behind(&self) -> Result> { let (_, wc) = self.working_copy()?; let mut base = wc.deps.clone(); base.sort(); let mut heads = self.bookmark_heads()?; heads.sort(); if base == heads || heads.is_empty() { return Ok(None); } let state = self.load_state_history(&heads)?; if !base.iter().all(|h| state.contains(h)) { return Ok(None); } let had = if base.is_empty() { 0 } else { self.load_state_history(&base)?.len() }; Ok(Some(state.len().saturating_sub(had)).filter(|n| *n > 0)) } /// Bring the working copy to `heads`, the state of its bookmark /// `bookmark`, when nothing would be lost: its base is part of that /// state, so the bookmark only adds to it. Clean, it is checked out; /// with uncommitted work, the work comes along when the new changes /// leave its paths alone (as `git pull` does), else it stays. Before, /// it followed only from exactly the bookmark's old heads, so a working /// copy already behind (the bookmark moved by `bookmark set`, a landing /// or a pull from another workspace) never caught up, and nothing said /// so (oxygen#74, #89). Returns whether it moved. pub fn follow_bookmark( &self, heads: &[ChangeHash], bookmark: &str, command: &str, ) -> Result { let snap = self.snapshot(command)?; let clean = snap.wc.tree_ops.is_empty() && snap.wc.edits.is_empty(); let mut base = snap.base.maximal(&snap.wc.deps); base.sort(); let mut target = heads.to_vec(); target.sort(); if base == target { return Ok(false); } let state = self.load_state_history(heads)?; if !base.iter().all(|h| state.contains(h)) { // The working copy holds changes the bookmark lacks: not ours to move. return Ok(false); } if clean { self.checkout(heads, false, &format!("{command} (checkout)"))?; Ok(true) } else { Ok(self .checkout_carry(heads, bookmark, &format!("{command} (checkout --carry)")) .is_ok()) } } /// Withdraw tag `tag` on `remote_name` (`oxy push --delete tags/NAME`, /// oxygenhub#301): withdrawn here first when it still is a tag here /// (`oxy tag delete`), then the withdrawal alone is sent, with its /// signer's identity and certificate. pub fn push_tag_withdrawal(&self, remote_name: &str, tag: &str) -> Result> { let remote = self.open_named_remote(remote_name)?; self.check_remote_id(remote.as_ref(), remote_name, true)?; let resp = remote.refs()?; let name = format!("tags/{tag}"); if resp.withdrawn.iter().any(|t| t.name == tag) { return Ok(vec![RefResult { name, ok: true, reason: "already withdrawn".into(), }]); } let Some(there) = resp.tags.iter().find(|t| t.name == tag) else { return fail(format!("{remote_name} has no tag {tag}")); }; if resp.policy.is_some() && self.secret_key()?.is_none() { return fail(format!("{remote_name} enforces a trust policy; withdrawing a tag needs an identity (`oxy identity create`)")); } if self.tag_withdrawal(tag)?.is_none() { match self.view()?.tags.get(tag) { Some(st) if *st == there.state => { self.tag_delete(tag, "")?; } Some(_) => { return fail(format!( "tag {tag} names another state here than on {remote_name}" )) } None => { return fail(format!( "there is no tag {tag} here to withdraw: `oxy fetch {remote_name}` first" )) } } } let Some(w) = self .withdrawn_refs()? .into_iter() .find(|w| w.name == tag && w.state == there.state) else { return fail(format!( "tag {tag} was withdrawn here at another state than {remote_name}'s" )); }; let mut bundle = Bundle::default(); let signers: Vec = w .attestation .iter() .flat_map(|a| a.sigs.iter().map(|s| s.key)) .collect(); for k in &signers { if let Ok(id) = self.identity(k) { bundle.objects.push((ObjectType::Identity, id.encode())); } } self.attach_certificates_for(&mut bundle, &signers)?; remote.push(bundle, vec![], None, vec![w]) } /// Push the current bookmark (spec §16.3), sealing this author's unsigned /// changes and attesting the reference when a key is available. /// Delete `bookmark` on a remote (spec §16.4): an update to the zero /// state, attested when we have an identity. The local bookmark, if /// any, is kept; only the remote-tracking entry goes. pub fn push_delete(&self, remote_name: &str, bookmark: &str) -> Result> { if let Some(tag) = bookmark.strip_prefix("tags/") { return self.push_tag_withdrawal(remote_name, tag); } let remote = self.open_named_remote(remote_name)?; self.check_remote_id(remote.as_ref(), remote_name, true)?; let resp = remote.refs()?; let Some(r) = resp.refs.iter().find(|r| r.name == bookmark) else { return fail(format!("{remote_name} has no bookmark {bookmark}")); }; if resp.policy.is_some() && self.secret_key()?.is_none() { return fail(format!("{remote_name} enforces a trust policy; deleting needs an identity (`oxy identity create`)")); } let attestation = match self.secret_key()? { Some(_) => Some(self.sign_attest(&Attest { repo: self.id()?, hlc: Hlc::now(), body: AttestBody::Ref { name: bookmark.to_string(), old_state: r.state, new_state: ObjectRef::ZERO, serial: r.last_serial.unwrap_or(0) + 1, }, })?), None => None, }; let results = remote.push( Bundle::default(), vec![RefUpdate { name: bookmark.to_string(), old_state: r.state, new_state: ObjectRef::ZERO, force: true, attestation, }], None, vec![], )?; if results.iter().any(|x| x.ok && x.name == bookmark) { let (op_ref, op) = self.head_op()?; let mut view = op.view.clone(); if let Some(m) = view.remotes.get_mut(remote_name) { m.remove(bookmark); } self.record_op( &[op_ref], &format!("oxy push {remote_name} --delete {bookmark}"), view, )?; } Ok(results) } /// Remove remote `name` (oxygen#86): every `remote.NAME.*` key of this /// repository's config (its URL, token, CA and store settings), its /// tracking bookmarks in the view, and a store remote's mirror under /// `.oxy/remotes/NAME`. Nothing stays to be read as if it were current. pub fn remove_remote(&mut self, name: &str) -> Result<()> { if !self.config.remotes.contains_key(name) { return fail(format!("no remote {name}: `oxy remote list` shows them")); } let prefix = format!("remote.{name}."); let keys: Vec = self .config .repo_layer() .keys() .filter(|k| k.starts_with(&prefix)) .cloned() .collect(); for k in keys { self.config.set(&k, "")?; } self.config.remotes.remove(name); self.save_config()?; let (op_ref, op) = self.head_op()?; if op.view.remotes.contains_key(name) { let mut view = op.view.clone(); view.remotes.remove(name); self.record_op(&[op_ref], &format!("oxy remote remove {name}"), view)?; } let mirror = self.oxy.join("remotes").join(name); if mirror.is_dir() { fs::remove_dir_all(&mirror)?; } Ok(()) } /// Send `bundle`'s blobs and chunks ahead of it, in parts of about /// `push.batch-bytes` of objects alone, when the whole would be larger /// (oxygen#133): one change can bring more blobs than one request may /// carry, and a proxy in front of a hub refuses a request past its limit /// (Cloudflare's is 100 MB). What is left, the changes with their /// identities and states, is returned to send as before. A receiver /// stores objects as they come, and checks a change's blobs are stored. fn send_objects_ahead( &self, remote: &dyn crate::protocol::Remote, remote_name: &str, mut bundle: Bundle, progress: crate::store::Progress, ) -> Result { let limit = self.config.push_batch_bytes; let changes: u64 = bundle .changes .iter() .map(|(c, _)| c.encode().len() as u64) .sum(); let objects: u64 = bundle.objects.iter().map(|(_, b)| b.len() as u64).sum(); if changes + objects <= limit { return Ok(bundle); } let (mut kept, mut parts, mut part, mut bytes) = (Vec::new(), Vec::new(), Vec::new(), 0u64); for (ty, body) in std::mem::take(&mut bundle.objects) { if !matches!(ty, ObjectType::Blob | ObjectType::Chunk) { kept.push((ty, body)); continue; } if !part.is_empty() && bytes + body.len() as u64 > limit { parts.push(std::mem::take(&mut part)); bytes = 0; } bytes += body.len() as u64; part.push((ty, body)); } if !part.is_empty() { parts.push(part); } let n = parts.len(); for (i, objects) in parts.into_iter().enumerate() { progress(&format!( "objects {} of {n}: {} blobs and chunks", i + 1, objects.len() )); let results = remote.push( Bundle { objects, ..Default::default() }, vec![], None, vec![], )?; match results.iter().find(|r| r.name == OBJECTS_RESULT) { Some(r) if r.ok => {} Some(r) => { return fail(format!( "{remote_name} refused objects part {} of {n}: {}", i + 1, r.reason )) } None => { return fail(format!( "{remote_name} does not take a push in parts, and this one is too large for one: update the server's oxy" )) } } } bundle.objects = kept; Ok(bundle) } /// `changes`, in dependency order, cut into batches of about /// `push.batch-bytes` by what each change and its texts and blobs weigh. /// There is always one, perhaps empty. fn push_batches(&self, changes: &[ChangeHash]) -> Result>> { let blob_len = |r: &ObjectRef| -> u64 { self.store .get_typed(r, ObjectType::Blob) .ok() .and_then(|b| Blob::decode(&b).ok()) .map_or(0, |b| b.length) }; let (mut out, mut batch, mut bytes) = (Vec::new(), Vec::new(), 0u64); for h in changes { let raw = self.store.get_typed(h, ObjectType::Change)?; let (c, texts) = Change::decode_skeleton(&raw, *h)?; let mut weight = raw.len() as u64; for t in &texts { if let crate::objects::TextRef::Blob(r, _, _) = t { weight += blob_len(r); } } for e in &c.edits { if let Edit::Replace { new: Some(b), .. } = e { weight += blob_len(b); } } if !batch.is_empty() && bytes + weight > self.config.push_batch_bytes { out.push(std::mem::take(&mut batch)); bytes = 0; } batch.push(*h); bytes += weight; } out.push(batch); Ok(out) } pub fn push(&self, remote_name: &str, force: bool) -> Result> { let bookmark = self.config.bookmark.clone(); self.push_bookmark(remote_name, &bookmark, force) } /// `push` for a named bookmark (`oxy push -b NAME`). pub fn push_bookmark( &self, remote_name: &str, bookmark: &str, force: bool, ) -> Result> { self.push_bookmark_with(remote_name, bookmark, force, &mut |_| {}) } /// `push_bookmark`, saying how a long push is going: one line per /// batch sent (oxygen#95). /// What a push may take the remote to hold already: the heads of the /// bookmark pushed, and of every other bookmark it lists that are held /// here, so a new bookmark made from main sends only what main lacks /// (oxygen#142), not its whole history again. pub(crate) fn remote_holds( &self, refs: &[crate::protocol::RemoteRef], bookmark_heads: &[ChangeHash], ) -> Vec { let mut have: Vec = bookmark_heads.to_vec(); for h in refs.iter().flat_map(|r| &r.heads) { if !have.contains(h) && self.store.has(h) { have.push(*h); } } have } pub fn push_bookmark_with( &self, remote_name: &str, bookmark: &str, force: bool, progress: crate::store::Progress, ) -> Result> { let (op_ref, op) = self.head_op()?; if !op.view.bookmarks.contains_key(bookmark) { return fail(format!("no bookmark {bookmark}")); } let bookmark = bookmark.to_string(); let local_heads = self.heads_of(&op.view, &bookmark)?; // pre-push hook: may veto before the remote is contacted. self.run_hook( "pre-push", &format!( "{{\"event\":\"pre-push\",\"remote\":{},\"bookmark\":{},\"heads\":{}}}", json_str(remote_name), json_str(&bookmark), json_array(local_heads.iter().map(|h| json_str(&h.to_hex()))) ), )?; let remote = self.open_named_remote(remote_name)?; self.check_remote_id(remote.as_ref(), remote_name, true)?; let resp = remote.refs()?; let local_policy = self.current_policy()?.map(|(_, sp)| sp); let resp_policy_serial = resp.policy.as_ref().map(|p| p.policy.serial); let policy_to_send = match (&local_policy, &resp.policy) { (Some(l), Some(r)) if l.policy.serial > r.policy.serial => Some(l.clone()), (Some(l), None) => Some(l.clone()), (Some(l), Some(r)) if l.policy.serial < r.policy.serial => { return fail(format!( "{remote_name} has a newer policy (serial {}); fetch first", r.policy.serial )); } _ => None, }; let remote_has_policy = resp.policy.is_some() || policy_to_send.is_some(); if remote_has_policy && self.secret_key()?.is_none() { return fail(format!("{remote_name} enforces a trust policy; pushing needs an identity (`oxy identity create`)")); } let remote_tags: BTreeSet = resp .tags .iter() .chain(&resp.withdrawn) .map(|t| t.name.clone()) .collect(); let tags_to_send: Vec = self .tag_refs()? .into_iter() .filter(|t| !remote_tags.contains(&t.name)) .collect(); // Tags withdrawn here that the remote still has, at the same state // (oxygenhub#301). let withdrawals_to_send: Vec = self .withdrawn_refs()? .into_iter() .filter(|w| { resp.tags .iter() .any(|t| t.name == w.name && t.state == w.state) }) .collect(); let refs = resp.refs; let empty = self.store.put_state(&StateObject::new(vec![]))?; let (remote_state, remote_heads, last_serial) = match refs.iter().find(|r| r.name == bookmark) { Some(r) => (r.state, r.heads.clone(), r.last_serial), None => (empty, vec![], None), }; let new_state = self .store .put_state(&StateObject::new(local_heads.clone()))?; // A push that carries a policy re-attests the bookmark even when its // state is unchanged, so the current attestation is always by a key // the current policy authorizes (a revoked key's attestation would // otherwise be refused by every later fetch). if new_state == remote_state && policy_to_send.is_none() && tags_to_send.is_empty() && withdrawals_to_send.is_empty() { return Ok(vec![RefResult { name: bookmark, ok: true, reason: "already up to date".into(), }]); } // A shallow repository knows the changes on its boundary without // holding them: a depth-0 clone's are the heads it cloned (#137). let boundary = self.boundary(); for h in &remote_heads { if !self.store.has(h) && !boundary.contains(h) { return fail(format!( "the remote has change {} which you do not; pull first (`oxy pull {remote_name}`), then push again", h.short(12) )); } } if !force { let closure = self.closure_in_store(&local_heads)?; if let Some(h) = remote_heads .iter() .find(|h| !closure.contains(h) && !boundary.contains(h)) { return fail(format!("not a fast-forward: remote head {} is not in your state; pull first or use --force", h.short(12))); } } let to_send = self.negotiate(&self.remote_holds(&refs, &remote_heads), &local_heads)?; let key = self.secret_key()?.map(|(k, _)| k); if let Some(key) = key { // Seal this author's changes that carry no valid authenticator, // and adopt anonymous ones (zero author, spec §14.4). The seals // are stored, so each batch's bundle carries those it needs. let mut unsigned = Vec::new(); for h in &to_send { let c = self.store.get_change_meta(h)?; if self.seals_for_author(&c.author, &key) && !matches!(self.authenticate_change(h)?, SignatureStatus::Valid { .. }) { unsigned.push(*h); } } for chunk in unsigned.chunks(crate::attest::MAX_SEAL_HASHES) { self.sign_attest(&Attest::seal(self.id()?, chunk.to_vec()))?; } } // A long history goes in batches, in dependency order, so neither // side holds it whole and no request passes the remote's limit: // all but the last carry objects alone; the last moves the bookmark // (oxygen#95). let batches = self.push_batches(&to_send)?; let (last, earlier) = batches.split_last().expect("at least one batch"); for (i, batch) in earlier.iter().enumerate() { progress(&format!( "part {} of {}: {} changes", i + 1, batches.len(), batch.len() )); let bundle = self.send_objects_ahead( remote.as_ref(), remote_name, self.bundle_for(batch)?, &mut *progress, )?; let results = remote.push(bundle, vec![], None, vec![])?; match results.iter().find(|r| r.name == OBJECTS_RESULT) { Some(r) if r.ok => {} Some(r) => { return fail(format!( "{remote_name} refused part {} of the push: {}", i + 1, r.reason )) } None => { return fail(format!( "{remote_name} does not take a push in parts, and this one is too long for one: update the server's oxy" )) } } } if batches.len() > 1 { progress(&format!( "part {} of {}: {} changes", batches.len(), batches.len(), last.len() )); } let mut bundle = self.send_objects_ahead( remote.as_ref(), remote_name, self.bundle_for(last)?, &mut *progress, )?; bundle.objects.push(( ObjectType::State, StateObject::new(local_heads.clone()).encode(), )); let attestation = match key { Some(_) => Some(self.sign_attest(&Attest { repo: self.id()?, hlc: Hlc::now(), body: AttestBody::Ref { name: bookmark.clone(), old_state: remote_state, new_state, serial: next_serial(last_serial), }, })?), None => None, }; // Whoever signed the move, and whom a delegate acts for, travel with // it: a receiver that has never seen a change by this delegate could // not otherwise check its signature (oxygenhub#129). if attestation.is_some() { for k in [self.signing_key(), self.config.author_key] .into_iter() .flatten() { if let Ok(id) = self.identity(&k) { let body = id.encode(); if !bundle .objects .iter() .any(|(t, b)| *t == ObjectType::Identity && *b == body) { bundle.objects.push((ObjectType::Identity, body)); } } } // A short-lived signing key's certificate goes with it (§14.7). let signers: Vec = self.signing_key().into_iter().collect(); self.attach_certificates_for(&mut bundle, &signers)?; } let serial = attestation.as_ref().and_then(|a| match a.attest.body { AttestBody::Ref { serial, .. } => Some(serial), _ => None, }); let sent_serial = policy_to_send.as_ref().map(|p| p.policy.serial); // Tag states must be present on the remote: bundle their changes too. for t in &tags_to_send { let extra = self.negotiate(&remote_heads, &t.heads)?; let extra: Vec = extra.into_iter().filter(|h| !to_send.contains(h)).collect(); if !extra.is_empty() { let more = self.bundle_for(&extra)?; bundle.changes.extend(more.changes); bundle.hashes.extend(more.hashes); bundle.attests.extend(more.attests); bundle.objects.extend(more.objects); } bundle.objects.push(( ObjectType::State, StateObject::new(t.heads.clone()).encode(), )); } // A tag's signer travels with it, as the bookmark's does: a tag made // by `oxy tag create` under a short-lived key was signed by a key the // receiver has never seen, so its identity and certificate go too. let tag_signers: Vec = tags_to_send .iter() .chain(&withdrawals_to_send) .filter_map(|t| t.attestation.as_ref()) .flat_map(|a| a.sigs.iter().map(|s| s.key)) .collect(); for k in &tag_signers { if let Ok(id) = self.identity(k) { let body = id.encode(); if !bundle .objects .iter() .any(|(t, b)| *t == ObjectType::Identity && *b == body) { bundle.objects.push((ObjectType::Identity, body)); } } } if !tag_signers.is_empty() { self.attach_certificates_for(&mut bundle, &tag_signers)?; } if let Some(l) = &policy_to_send { let remote_serial = resp_policy_serial; bundle.policies = self .policy_history()? .into_iter() .filter(|sp| { remote_serial.is_none_or(|r| sp.policy.serial > r) && sp.policy.serial < l.policy.serial }) .collect(); } let results = remote.push( bundle, vec![RefUpdate { name: bookmark.clone(), old_state: remote_state, new_state, force, attestation, }], policy_to_send, tags_to_send .into_iter() .chain(withdrawals_to_send) .collect(), )?; if let (Some(serial), true) = ( sent_serial, results.iter().any(|r| r.ok && r.name == "policy"), ) { self.record_policy_serial(remote_name, serial)?; } if results.iter().any(|r| r.ok && r.name == bookmark) { let mut view = op.view.clone(); view.remotes .entry(remote_name.to_string()) .or_default() .insert(bookmark.clone(), new_state); self.record_op(&[op_ref], &format!("oxy push {remote_name}"), view)?; if let Some(s) = serial { self.record_serial(remote_name, &bookmark, s)?; } } Ok(results) } /// Clone a remote into `dest`: same repository ID, `origin` remote, /// bookmark set to the remote's, working copy checked out. pub fn clone(url: &str, dest: &Path) -> Result { Self::clone_with(url, dest, None, &[]) } /// Clone, leaving blobs above `max_blob` bytes to be fetched on first /// read, and materializing only `sparse` paths (which implies lazy blobs /// unless a size is given, so out-of-scope binaries are never fetched). pub fn clone_with( url: &str, dest: &Path, max_blob: Option, sparse: &[String], ) -> Result { Self::clone_full(url, dest, max_blob, sparse, None) } /// Shallow clone (spec §17): a checkpoint of the bookmark's heads plus the /// changes within `depth` steps of them; the rest is the boundary. pub fn clone_shallow( url: &str, dest: &Path, depth: u32, max_blob: Option, sparse: &[String], ) -> Result { Self::clone_full(url, dest, max_blob, sparse, Some(depth)) } fn clone_full( url: &str, dest: &Path, max_blob: Option, sparse: &[String], depth: Option, ) -> Result { Ok(Self::clone_reporting(url, dest, max_blob, sparse, depth)?.0) } /// Clone as [`Repo::clone_with`] (or [`Repo::clone_shallow`] with a /// `depth`), also returning the bookmarks the fetch left out because /// their attestation could not be verified ([`FetchReport::skipped`]), /// for the caller to warn about. The bookmark checked out is never one /// of them: that fails the clone. pub fn clone_reporting( url: &str, dest: &Path, max_blob: Option, sparse: &[String], depth: Option, ) -> Result<(Repo, Vec<(String, String)>)> { let options = crate::credentials::Credentials::user().options_for(url); Self::clone_using(url, dest, max_blob, sparse, depth, &options) } /// Clone a subrepo with the credentials the parent would use for its /// hub (see [`Repo::credentials_for`]). Nothing is recorded in the /// subrepo: it finds the same ones again through its parent. pub(crate) fn clone_as_subrepo(&self, url: &str, dest: &Path) -> Result { let options = self.credentials_for(url); Ok(Self::clone_using(url, dest, None, &[], None, &options)?.0) } /// The name of a remote of this repository on the same hub (scheme, /// host and port) as `url`. fn remote_on_same_hub(&self, url: &str) -> Option { let origin = |u: &str| -> Option { let (scheme, rest) = u.split_once("://")?; if !matches!(scheme, "http" | "https") { return None; } let host = rest.split('/').next()?; Some(format!( "{}://{}", scheme.to_ascii_lowercase(), host.to_ascii_lowercase() )) }; let want = origin(url)?; self.config .remotes .iter() .find(|(_, u)| origin(u).as_deref() == Some(want.as_str())) .map(|(name, _)| name.clone()) } /// The token and CA to reach `url` with: those of a remote of this /// repository on the same hub (scheme, host and port), else the /// environment's (`OXY_TOKEN`, `OXY_CA`), the user's credential store /// and enclosing checkouts', as a clone would use them. pub(crate) fn credentials_for(&self, url: &str) -> RemoteOptions { match self.remote_on_same_hub(url) { Some(name) => RemoteOptions { repo_id: None, cache_dir: None, ..self.remote_options(&name) }, None => { let mut options = RemoteOptions::default(); self.fill_credentials(url, &mut options); options } } } fn clone_using( url: &str, dest: &Path, max_blob: Option, sparse: &[String], depth: Option, options: &RemoteOptions, ) -> Result<(Repo, Vec<(String, String)>)> { let remote = open_remote(url, options)?; let id = remote.repo_id()?; if id == crate::protocol::UNBOUND_ID { return fail(format!( "{url} is empty and unbound: push a repository to it first" )); } let mut repo = Repo::init_with_id(dest, id)?; repo.config .set("remote.origin.url", &crate::protocol::stored_url(url))?; if let Some(m) = max_blob { repo.config.set("filter.max-blob", &m.to_string())?; } else if !sparse.is_empty() { repo.config.set("filter.max-blob", "0")?; } if !sparse.is_empty() { repo.config.set("filter.paths", &sparse.join(","))?; } repo.save_config()?; let skipped = if let Some(depth) = depth { repo.fetch_shallow("origin", depth)?; Vec::new() } else { repo.fetch("origin")?.skipped }; let (op_ref, op) = repo.head_op()?; // The remote's default bookmark, else ours (main), else its only // one; a bookmark the fetch skipped counts, so the clone fails // rather than check out another (oxygenhub#266). let mut theirs = op.view.remotes.get("origin").cloned().unwrap_or_default(); for (name, _) in &skipped { theirs.insert(name.clone(), ObjectRef::ZERO); } // Asked of the remote as now configured, so a store-backed one // answers from the mirror the fetch just filled. let advertised = repo .open_named_remote("origin") .ok() .and_then(|r| r.refs().ok()) .and_then(|r| r.default_bookmark); let bookmark = advertised .filter(|b| theirs.contains_key(b)) .or_else(|| { theirs .contains_key(&repo.config.bookmark) .then(|| repo.config.bookmark.clone()) }) .or_else(|| { (theirs.len() == 1) .then(|| theirs.keys().next().cloned()) .flatten() }) .unwrap_or_else(|| repo.config.bookmark.clone()); if let Some((_, why)) = skipped.iter().find(|(name, _)| *name == bookmark) { return fail(format!("cannot check out {bookmark}: {why}")); } let initial = repo.config.bookmark.clone(); if bookmark != repo.config.bookmark { fs::create_dir_all(repo.workspace_bookmark_path().parent().unwrap())?; write_atomic(&repo.workspace_bookmark_path(), bookmark.as_bytes())?; repo.config.bookmark = bookmark.clone(); } if let Some(state) = theirs.get(&bookmark).copied() { let heads = repo.store.get_state(&state)?.heads; let mut view = op.view.clone(); // The empty bookmark `init` made goes when another is checked out. if initial != bookmark && repo.heads_of(&view, &initial)?.is_empty() { view.bookmarks.remove(&initial); } view.bookmarks.insert(bookmark, state); repo.record_op(&[op_ref], &format!("oxy clone {url}"), view)?; repo.checkout(&heads, false, "oxy clone (checkout)")?; } Ok((repo, skipped)) } } /// The serial of a reference's next attestation: one more than the last the /// remote reports, or, for a bookmark the remote does not have, the clock in /// milliseconds, so a bookmark created again after a deletion goes past the /// deletion's serial the remote keeps. fn next_serial(last: Option) -> u64 { match last { Some(l) => l + 1, None => Hlc::now().wall_ms.max(1), } }