Remotes: local transport, push with seals and reference attestations, fetch, pull as state union, clone 5b9c6ad45f6e

//! Attestations (spec §14.4): signed statements about references and batches

//! of changes. Only the `ref` and `seal` kinds exist so far.

use crate::cbor::{self, Value};

use crate::hash::{hash_object, ChangeHash, ObjectRef, ObjectType};

use crate::objects::{array, bytes, check_version, field, map, oref, text, uint, Hlc, ObjectError, FORMAT_VERSION};

pub const MAX_SEAL_HASHES: usize = 4096;

#[derive(Clone, Debug, PartialEq, Eq)]

pub enum AttestBody {

    /// A bookmark moved from `old_state` to `new_state` (spec §10.2).

    Ref { name: String, old_state: ObjectRef, new_state: ObjectRef, serial: u64 },

    /// One signature covering a batch of changes (spec §14.4).

    Seal { hashes: Vec<ChangeHash> },

}

#[derive(Clone, Debug, PartialEq, Eq)]

pub struct Attest {

    /// Repository ID chosen at init.

    pub repo: [u8; 16],

    pub hlc: Hlc,

    pub body: AttestBody,

}

impl Attest {

    pub fn kind(&self) -> &'static str {

        match self.body {

            AttestBody::Ref { .. } => "ref",

            AttestBody::Seal { .. } => "seal",

        }

    }

    pub fn seal(repo: [u8; 16], mut hashes: Vec<ChangeHash>) -> Attest {

        hashes.sort();

        hashes.dedup();

        Attest { repo, hlc: Hlc::now(), body: AttestBody::Seal { hashes } }

    }

    pub fn to_value(&self) -> Value {

        let body = match &self.body {

            AttestBody::Ref { name, old_state, new_state, serial } => Value::map([

                (0, Value::Text(name.clone())),

                (1, Value::Bytes(old_state.0.to_vec())),

                (2, Value::Bytes(new_state.0.to_vec())),

                (3, Value::Uint(*serial)),

            ]),

            AttestBody::Seal { hashes } => {

                Value::map([(0, Value::Array(hashes.iter().map(|h| Value::Bytes(h.0.to_vec())).collect()))])

            }

        };

        Value::map([

            (0, Value::Uint(FORMAT_VERSION)),

            (1, Value::Text(self.kind().into())),

            (2, Value::Bytes(self.repo.to_vec())),

            (3, self.hlc.to_value()),

            (4, body),

        ])

    }

    pub fn encode(&self) -> Vec<u8> {

        cbor::encode(&self.to_value())

    }

    pub fn hash(&self) -> ObjectRef {

        hash_object(ObjectType::Attest, &self.encode())

    }

    pub fn decode(b: &[u8]) -> Result<Attest, ObjectError> {

        let v = cbor::decode(b)?;

        let m = map(&v, "attest")?;

        check_version(m)?;

        let kind = text(field(m, 1, "kind")?, "kind")?;

        let repo: [u8; 16] = bytes(field(m, 2, "repo")?, "repo")?

            .try_into()

            .map_err(|_| ObjectError::Schema("repo: expected 16 bytes".into()))?;

        let hlc = Hlc::from_value(field(m, 3, "hlc")?)?;

        let b = map(field(m, 4, "body")?, "body")?;

        let body = match kind {

            "ref" => AttestBody::Ref {

                name: text(field(b, 0, "name")?, "name")?.to_owned(),

                old_state: oref(field(b, 1, "old_state")?, "old_state")?,

                new_state: oref(field(b, 2, "new_state")?, "new_state")?,

                serial: uint(field(b, 3, "serial")?, "serial")?,

            },

            "seal" => {

                let hashes = array(field(b, 0, "hashes")?, "hashes")?.iter().map(|h| oref(h, "hash")).collect::<Result<Vec<_>, _>>()?;

                if hashes.is_empty() || hashes.len() > MAX_SEAL_HASHES || !hashes.windows(2).all(|w| w[0] < w[1]) {

                    return Err(ObjectError::Schema("seal: hashes must be 1..4096, sorted and unique".into()));

                }

                AttestBody::Seal { hashes }

            }

            other => return Err(ObjectError::Schema(format!("unknown attestation kind {other:?}"))),

        };

        Ok(Attest { repo, hlc, body })

    }

}

#[cfg(test)]

mod tests {

    use super::*;

    #[test]

    fn round_trips() {

        let r = Attest {

            repo: [1; 16],

            hlc: Hlc { wall_ms: 3, logical: 0 },

            body: AttestBody::Ref { name: "main".into(), old_state: ObjectRef([1; 32]), new_state: ObjectRef([2; 32]), serial: 7 },

        };

        assert_eq!(Attest::decode(&r.encode()).unwrap(), r);

        assert_eq!(r.kind(), "ref");

        let s = Attest::seal([1; 16], vec![ObjectRef([9; 32]), ObjectRef([3; 32]), ObjectRef([9; 32])]);

        let d = Attest::decode(&s.encode()).unwrap();

        assert!(matches!(&d.body, AttestBody::Seal { hashes } if hashes == &vec![ObjectRef([3; 32]), ObjectRef([9; 32])]));

        assert_ne!(s.hash(), r.hash());

    }

    #[test]

    fn rejects_bad_seals() {

        let empty = Attest { repo: [0; 16], hlc: Hlc::default(), body: AttestBody::Seal { hashes: vec![] } };

        assert!(Attest::decode(&empty.encode()).is_err());

        let unsorted = Attest { repo: [0; 16], hlc: Hlc::default(), body: AttestBody::Seal { hashes: vec![ObjectRef([2; 32]), ObjectRef([1; 32])] } };

        assert!(Attest::decode(&unsorted.encode()).is_err());

    }

}

//! 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; an HTTPS transport implements the same trait and the

//! server side reuses `Repo::receive_push` and friends unchanged.

use crate::attest::Attest;

use crate::hash::{ChangeHash, ObjectRef, ObjectType};

use crate::identity::Signature;

use crate::objects::Change;

use crate::repo::{Repo, RepoError, Result};

use std::path::Path;

/// An attestation together with the signatures over its ObjectRef.

#[derive(Clone, Debug, PartialEq, Eq)]

pub struct SignedAttest {

    pub attest: Attest,

    pub sigs: Vec<Signature>,

}

/// One reference as a remote reports it.

#[derive(Clone, Debug)]

pub struct RemoteRef {

    pub name: String,

    pub state: ObjectRef,

    pub heads: Vec<ChangeHash>,

    pub attestation: Option<SignedAttest>,

}

/// Everything needed to add a set of changes to a repository.

#[derive(Clone, Debug, Default)]

pub struct Bundle {

    /// In dependency order, each with its individual signatures.

    pub changes: Vec<(Change, Vec<Signature>)>,

    /// Seals covering the changes.

    pub attests: Vec<SignedAttest>,

    /// Chunks, blobs, identities, and state objects referenced by the changes.

    pub objects: Vec<(ObjectType, Vec<u8>)>,

}

impl Bundle {

    pub fn is_empty(&self) -> bool {

        self.changes.is_empty() && self.attests.is_empty() && self.objects.is_empty()

    }

}

/// A requested bookmark move (spec §10.2, §16.3).

#[derive(Clone, Debug)]

pub struct RefUpdate {

    pub name: String,

    pub old_state: ObjectRef,

    pub new_state: ObjectRef,

    pub force: bool,

    pub attestation: Option<SignedAttest>,

}

#[derive(Clone, Debug, PartialEq, Eq)]

pub struct RefResult {

    pub name: String,

    pub ok: bool,

    pub reason: String,

}

pub trait Remote {

    fn repo_id(&self) -> Result<[u8; 16]>;

    fn refs(&self) -> Result<Vec<RemoteRef>>;

    /// Changes in the closure of `want` but not of `have`, in dependency order.

    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>>;

    fn fetch(&self, changes: &[ChangeHash]) -> Result<Bundle>;

    fn push(&self, bundle: Bundle, updates: Vec<RefUpdate>) -> Result<Vec<RefResult>>;

}

/// Another repository on this machine, served in-process.

pub struct LocalRemote {

    pub repo: Repo,

}

impl LocalRemote {

    pub fn open(path: &Path) -> Result<LocalRemote> {

        Ok(LocalRemote { repo: Repo::open(path)? })

    }

}

impl Remote for LocalRemote {

    fn repo_id(&self) -> Result<[u8; 16]> {

        self.repo.id()

    }

    fn refs(&self) -> Result<Vec<RemoteRef>> {

        self.repo.remote_refs()

    }

    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>> {

        self.repo.negotiate(have, want)

    }

    fn fetch(&self, changes: &[ChangeHash]) -> Result<Bundle> {

        self.repo.bundle_for(changes)

    }

    fn push(&self, bundle: Bundle, updates: Vec<RefUpdate>) -> Result<Vec<RefResult>> {

        self.repo.receive_push(bundle, updates)

    }

}

/// Open a remote by URL. Only local paths (optionally `file://`) exist so far.

pub fn open_remote(url: &str) -> Result<Box<dyn Remote>> {

    let path = url.strip_prefix("file://").unwrap_or(url);

    if path.contains("://") {

        return Err(RepoError::Msg(format!("unsupported remote URL {url:?}: only local paths are implemented")));

    }

    Ok(Box::new(LocalRemote::open(Path::new(path))?))

}

pub mod attest;

pub mod protocol;

    /// `remotes/<remote>/bookmarks/<name>`: remote → bookmark → state ObjectRef.

    pub remotes: BTreeMap<String, BTreeMap<String, ObjectRef>>,

            (

                2,

                Value::Array(

                    self.remotes

                        .iter()

                        .map(|(r, bms)| {

                            Value::Array(vec![

                                Value::Text(r.clone()),

                                Value::Array(

                                    bms.iter().map(|(n, st)| Value::Array(vec![Value::Text(n.clone()), Value::Bytes(st.0.to_vec())])).collect(),

                                ),

                            ])

                        })

                        .collect(),

                ),

            ),

        if let Some(rs) = m.get(&2) {

            for r in array(rs, "remotes")? {

                let pair = array(r, "remote")?;

                if pair.len() != 2 {

                    return Err(ObjectError::Schema("remote: expected [name, bookmarks]".into()));

                }

                let name = text(&pair[0], "remote name")?.to_owned();

                let mut bms = BTreeMap::new();

                for b in array(&pair[1], "remote bookmarks")? {

                    let bp = array(b, "remote bookmark")?;

                    if bp.len() != 2 {

                        return Err(ObjectError::Schema("remote bookmark: expected [name, state]".into()));

                    }

                    bms.insert(text(&bp[0], "bookmark name")?.to_owned(), oref(&bp[1], "bookmark state")?);

                }

                view.remotes.insert(name, bms);

            }

        }

        view.remotes.entry("origin".into()).or_default().insert("main".into(), ObjectRef([4; 32]));

//! Not yet implemented from the spec: trust policy (every remote is

//! "unverified", so unsigned pushes are accepted), packs, checkpoints,

//! multiple workspaces, and network transports (only local paths). Changes are signed individually when `core.sign` is

use crate::attest::{Attest, AttestBody};

use crate::protocol::{open_remote, Bundle, RefResult, RefUpdate, RemoteRef, SignedAttest};

use rand::RngCore;

    /// `remote.<name>.url`.

    pub remotes: BTreeMap<String, String>,

            remotes: BTreeMap::new(),

            k if k.starts_with("remote.") && k.ends_with(".url") => {

                let name = &k["remote.".len()..k.len() - ".url".len()];

                if name.is_empty() || name.contains('/') {

                    return fail(format!("invalid remote name in {key:?}"));

                }

                if value.is_empty() {

                    self.remotes.remove(name);

                } else {

                    self.remotes.insert(name.to_string(), value.into());

                }

            }

        let mut text = format!(

        for (name, url) in &self.remotes {

            text.push_str(&format!("remote.{name}.url = \"{url}\"\n"));

        }

    Valid { key: KeyId, name: String, email: String, sealed: bool },

        let mut id = [0u8; 16];

        rand::rngs::OsRng.fill_bytes(&mut id);

        Self::init_with_id(root, id)

    }

    /// Initialize a repository with a given repository ID (clones share the

    /// ID of their origin, spec §14.4).

    pub fn init_with_id(root: &Path, id: [u8; 16]) -> Result<Repo> {

        fs::create_dir_all(root)?;

        write_atomic(&oxy.join("id"), hex::encode(id).as_bytes())?;

    /// The repository ID (spec §14.4). Repositories made before IDs existed get one on first use.

    pub fn id(&self) -> Result<[u8; 16]> {

        let path = self.oxy.join("id");

        match fs::read_to_string(&path) {

            Ok(h) => hex::decode(h.trim()).ok().and_then(|v| v.try_into().ok()).ok_or_else(|| RepoError::Msg("corrupt .oxy/id".into())),

            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {

                let mut id = [0u8; 16];

                rand::rngs::OsRng.fill_bytes(&mut id);

                write_atomic(&path, hex::encode(id).as_bytes())?;

                Ok(id)

            }

            Err(e) => Err(e.into()),

        }

    }

            match self.check_signer(&change, sig, Domain::Change, hash) {

                Ok(id) => return Ok(SignatureStatus::Valid { key: sig.key, name: id.name, email: id.email, sealed: false }),

                Err(e) => last_err = e,

            }

        }

        Ok(SignatureStatus::Invalid(last_err))

    }

    /// Verify `sig` over `target` and check the author-or-delegate rule for `change`.

    fn check_signer(&self, change: &Change, sig: &Signature, domain: Domain, target: &ObjectRef) -> std::result::Result<Identity, String> {

        let id = self.store.get_identity(&sig.key).map_err(|_| format!("unknown key {}", sig.key.short(12)))?;

        let authorised =

            sig.key == change.author || self.store.get_identity(&change.author).map(|a| a.delegates.contains(&sig.key)).unwrap_or(false);

        if !authorised {

            return Err(format!("key {} is neither the author nor a delegate", sig.key.short(12)));

        }

        identity::verify(&id, sig, domain, target).map_err(|e| e.to_string())?;

        Ok(id)

    }

    /// Like `verify_change`, but a change with no valid individual signature

    /// is also authenticated by a valid seal covering it (spec §14.4).

    pub fn authenticate_change(&self, hash: &ChangeHash) -> Result<SignatureStatus> {

        let direct = self.verify_change(hash)?;

        if matches!(direct, SignatureStatus::Valid { .. }) {

            return Ok(direct);

        }

        let change = self.store.get_change(hash)?;

        let mut last_err = match &direct {

            SignatureStatus::Invalid(e) => e.clone(),

            _ => String::new(),

        };

        for seal_ref in self.store.seals_for(hash)? {

            let attest = self.store.get_attest(&seal_ref)?;

            let AttestBody::Seal { hashes } = &attest.body else { continue };

            if !hashes.contains(hash) {

                continue;

            }

            for sig in self.store.signatures(&seal_ref)? {

                match self.check_signer(&change, &sig, Domain::Attest, &seal_ref) {

                    Ok(id) => return Ok(SignatureStatus::Valid { key: sig.key, name: id.name, email: id.email, sealed: true }),

                    Err(e) => last_err = e,

                }

            }

        }

        if last_err.is_empty() { Ok(SignatureStatus::Unsigned) } else { Ok(SignatureStatus::Invalid(last_err)) }

    }

    /// Store and sign an attestation with the configured key.

    fn sign_attest(&self, attest: &Attest) -> Result<SignedAttest> {

        let Some((key, sk)) = self.secret_key()? else {

            return fail("no secret key: run `oxy identity create` first");

        };

        let r = self.store.put_attest(attest)?;

        let sig = sk.sign(Domain::Attest, &key, &r);

        self.store.put_signature(&r, &sig)?;

        if let AttestBody::Seal { hashes } = &attest.body {

            for h in hashes {

                self.store.index_seal(h, &r)?;

            }

        }

        Ok(SignedAttest { attest: attest.clone(), sigs: vec![sig] })

    }

    // ------------------------------------------------------------ serving (spec §16)

    fn ref_attest_path(&self, bookmark: &str) -> PathBuf {

        self.oxy.join("attest").join(bookmark)

    }

    /// The latest reference attestation stored for a bookmark.

    pub fn latest_ref_attest(&self, bookmark: &str) -> Result<Option<(ObjectRef, SignedAttest)>> {

        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()),

        }

    }

    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<Vec<RemoteRef>> {

        let view = self.view()?;

        let mut out = Vec::new();

        for (name, state) in &view.bookmarks {

            out.push(RemoteRef {

                name: name.clone(),

                state: *state,

                heads: self.store.get_state(state)?.heads,

                attestation: self.latest_ref_attest(name)?.map(|(_, a)| a),

            });

        }

        Ok(out)

    }

    /// Every change reachable from `heads` that is in the store.

    fn closure_in_store(&self, heads: &[ChangeHash]) -> Result<BTreeSet<ChangeHash>> {

        let mut seen = BTreeSet::new();

        let mut stack: Vec<ChangeHash> = 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(&h)?.deps);

        }

        Ok(seen)

    }

    /// `closure(want) \ closure(have)`, dependencies first (spec §16.2).

    pub fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>> {

        let have = self.closure_in_store(have)?;

        let mut out = Vec::new();

        let mut done: BTreeSet<ChangeHash> = have.clone();

        // Iterative post-order DFS.

        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) {

                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(&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<Bundle> {

        let mut bundle = Bundle::default();

        let mut objects: BTreeSet<ObjectRef> = BTreeSet::new();

        let mut seals: BTreeSet<ObjectRef> = BTreeSet::new();

        let add_object = |r: &ObjectRef, objects: &mut BTreeSet<ObjectRef>| {

            if !r.is_zero() && self.store.has(r) {

                objects.insert(*r);

            }

        };

        for h in changes {

            let c = self.store.get_change(h)?;

            let sigs = self.store.signatures(h)?;

            add_object(&c.author, &mut objects);

            for s in &sigs {

                add_object(&s.key, &mut objects);

            }

            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) {

                            objects.insert(*b);

                            if let Ok(blob) = Blob::decode(&self.store.get_typed(b, ObjectType::Blob)?) {

                                objects.extend(blob.chunks.iter().copied());

                            }

                        }

                    }

                }

            }

            for s in self.store.seals_for(h)? {

                seals.insert(s);

            }

            bundle.changes.push((c, sigs));

        }

        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 });

        }

        for r in objects {

            let (ty, body) = self.store.get(&r)?;

            bundle.objects.push((ty, body));

        }

        Ok(bundle)

    }

    /// 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<Vec<ChangeHash>> {

        for (ty, body) in &bundle.objects {

            match ty {

                ObjectType::Chunk | ObjectType::Blob | ObjectType::Identity | ObjectType::State => {

                    self.store.put(*ty, body)?;

                }

                other => return fail(format!("bundle may not carry {} objects", other.name())),

            }

        }

        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)?;

                }

            }

        }

        let incoming: BTreeSet<ChangeHash> = bundle.changes.iter().map(|(c, _)| c.hash()).collect();

        let mut imported = Vec::new();

        for (c, sigs) in &bundle.changes {

            let h = c.hash();

            for d in &c.deps {

                if !self.store.has(d) && !incoming.contains(d) {

                    return fail(format!("change {} depends on {}, which is neither stored nor in the bundle", h.short(12), d.short(12)));

                }

            }

            self.store.put_change(c)?;

            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.

    fn check_ref_attest(&self, sa: &SignedAttest, name: &str, old_state: &ObjectRef, new_state: &ObjectRef, last_serial: Option<u64>) -> std::result::Result<u64, String> {

        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 {

                return Err(format!("attestation serial {serial} is not greater than the last seen {last}"));

            }

        }

        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)))?;

        identity::verify(&id, sig, Domain::Attest, &r).map_err(|e| e.to_string())?;

        Ok(*serial)

    }

    /// Accept a push (spec §16.3). Without a trust policy every repository

    /// is unverified, so unsigned changes and unattested updates are accepted;

    /// attestations that are present must still be valid.

    pub fn receive_push(&self, bundle: Bundle, updates: Vec<RefUpdate>) -> Result<Vec<RefResult>> {

        self.import_bundle(&bundle, false)?;

        let (op_ref, op) = self.head_op()?;

        let mut view = op.view.clone();

        let mut results = Vec::new();

        let mut names = Vec::new();

        for u in updates {

            let outcome = self.apply_update(&mut view, &u);

            let (ok, reason) = match outcome {

                Ok(()) => (true, String::new()),

                Err(e) => (false, e),

            if ok {

                names.push(u.name.clone());

            }

            results.push(RefResult { name: u.name, ok, reason });

        }

        if !names.is_empty() {

            self.record_op(&[op_ref], &format!("push received: {}", names.join(" ")), view)?;

        }

        Ok(results)

    }

    fn apply_update(&self, view: &mut View, u: &RefUpdate) -> std::result::Result<(), String> {

        let empty = self.store.put_state(&StateObject::new(vec![])).map_err(|e| e.to_string())?;

        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(&current).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(sa) = &u.attestation {

            let last = self.latest_ref_attest(&u.name).map_err(|e| e.to_string())?.and_then(|(_, a)| match a.attest.body {

                AttestBody::Ref { serial, .. } => Some(serial),

                _ => None,

            });

            self.check_ref_attest(sa, &u.name, &u.old_state, &u.new_state, last)?;

            let r = self.store.put_attest(&sa.attest).map_err(|e| e.to_string())?;

            for sig in &sa.sigs {

                self.store.put_signature(&r, sig).map_err(|e| e.to_string())?;

            }

            self.set_ref_attest(&u.name, &r).map_err(|e| e.to_string())?;

        }

        view.bookmarks.insert(u.name.clone(), u.new_state);

        Ok(())

    }

    // ------------------------------------------------------------ client side

    fn remote_url(&self, name: &str) -> Result<String> {

        self.config.remotes.get(name).cloned().ok_or_else(|| RepoError::Msg(format!("no remote named {name:?}; add one with `oxy remote add`")))

    }

    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<u64> {

        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())

    }

    /// Every head this repository knows: bookmarks, remote tracking, working copy.

    fn known_heads(&self, view: &View) -> Result<Vec<ChangeHash>> {

        let mut have: BTreeSet<ChangeHash> = 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/<name>/bookmarks/*` (spec §10.2).

    pub fn fetch(&self, remote_name: &str) -> Result<Vec<(String, Vec<ChangeHash>, usize)>> {

        let remote = open_remote(&self.remote_url(remote_name)?)?;

        if remote.repo_id()? != self.id()? {

            return fail(format!("remote {remote_name:?} is a different repository (repository IDs differ)"));

        }

        let (op_ref, op) = self.head_op()?;

        let mut view = op.view.clone();

        let have = self.known_heads(&view)?;

        let mut report = Vec::new();

        for r in remote.refs()? {

            let needed = remote.negotiate(&have, &r.heads)?;

            let bundle = remote.fetch(&needed)?;

            let imported = self.import_bundle(&bundle, false)?;

            for h in &r.heads {

                if !self.store.has(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(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(sa, &r.name, &old_state, &new_state, self.last_seen_serial(remote_name, &r.name))

                    .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));

                }

                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);

            report.push((r.name, r.heads, imported.len()));

        }

        if view != op.view {

            self.record_op(&[op_ref], &format!("oxy fetch {remote_name}"), view)?;

        }

        Ok(report)

    }

    /// 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<PullReport> {

        self.fetch(remote_name)?;

        let (op_ref, op) = self.head_op()?;

        let bookmark = self.config.bookmark.clone();

        let local = self.heads_of(&op.view, &bookmark)?;

        let Some(remote_state) = op.view.remotes.get(remote_name).and_then(|b| b.get(&bookmark)).copied() else {

            return Ok(PullReport { heads: local, changed: false, checked_out: false });

        };

        let remote_heads = self.store.get_state(&remote_state)?.heads;

        let mut all = local.clone();

        all.extend(remote_heads);

        let st = self.load_state(&all)?;

        let mut heads = st.maximal(&all);

        heads.sort();

        let mut sorted_local = local.clone();

        sorted_local.sort();

        if heads == sorted_local {

            return Ok(PullReport { heads, changed: false, checked_out: false });

        }

        let mut view = op.view.clone();

        view.bookmarks.insert(bookmark.clone(), self.store.put_state(&StateObject::new(heads.clone()))?);

        self.record_op(&[op_ref], &format!("oxy pull {remote_name}"), view)?;

        // Follow along if the working copy was clean and on the old heads.

        let snap = self.snapshot("oxy pull")?;

        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 checked_out = false;

        if clean && base == sorted_local {

            self.checkout(&heads, false, "oxy pull (checkout)")?;

            checked_out = true;

        }

        Ok(PullReport { heads, changed: true, checked_out })

    }

    /// Push the current bookmark (spec §16.3), sealing this author's unsigned

    /// changes and attesting the reference when a key is available.

    pub fn push(&self, remote_name: &str, force: bool) -> Result<Vec<RefResult>> {

        let remote = open_remote(&self.remote_url(remote_name)?)?;

        if remote.repo_id()? != self.id()? {

            return fail(format!("remote {remote_name:?} is a different repository (repository IDs differ)"));

        }

        let (op_ref, op) = self.head_op()?;

        let bookmark = self.config.bookmark.clone();

        let local_heads = self.heads_of(&op.view, &bookmark)?;

        let refs = remote.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.attestation.as_ref().and_then(|a| match a.attest.body {

                    AttestBody::Ref { serial, .. } => Some(serial),

                    _ => None,

                }),

            ),

            None => (empty, vec![], None),

        };

        let new_state = self.store.put_state(&StateObject::new(local_heads.clone()))?;

        if new_state == remote_state {

            return Ok(vec![RefResult { name: bookmark, ok: true, reason: "already up to date".into() }]);

        }

        for h in &remote_heads {

            if !self.store.has(h) {

                return fail(format!("the remote has change {} which you do not; pull first", 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)) {

                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(&remote_heads, &local_heads)?;

        let mut bundle = self.bundle_for(&to_send)?;

        let key = self.secret_key()?.map(|(k, _)| k);

        if let Some(key) = key {

            // Seal this author's changes that carry no valid authenticator.

            let mut unsigned = Vec::new();

            for h in &to_send {

                let c = self.store.get_change(h)?;

                if c.author == key && !matches!(self.authenticate_change(h)?, SignatureStatus::Valid { .. }) {

                    unsigned.push(*h);

                }

            }

            for chunk in unsigned.chunks(crate::attest::MAX_SEAL_HASHES) {

                let sa = self.sign_attest(&Attest::seal(self.id()?, chunk.to_vec()))?;

                bundle.attests.push(sa);

            bundle = self.bundle_for(&to_send)?;

        }

        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: last_serial.unwrap_or(0) + 1 },

            })?),

            None => None,

        };

        let serial = attestation.as_ref().and_then(|a| match a.attest.body {

            AttestBody::Ref { serial, .. } => Some(serial),

            _ => None,

        });

        let results = remote.push(bundle, vec![RefUpdate { name: bookmark.clone(), old_state: remote_state, new_state, force, attestation }])?;

        if results.iter().any(|r| r.ok) {

            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<Repo> {

        let remote = open_remote(url)?;

        let id = remote.repo_id()?;

        let mut repo = Repo::init_with_id(dest, id)?;

        repo.config.remotes.insert("origin".into(), url.to_string());

        repo.save_config()?;

        repo.fetch("origin")?;

        let (op_ref, op) = repo.head_op()?;

        let bookmark = repo.config.bookmark.clone();

        if let Some(state) = op.view.remotes.get("origin").and_then(|b| b.get(&bookmark)).copied() {

            let heads = repo.store.get_state(&state)?.heads;

            let mut view = op.view.clone();

            view.bookmarks.insert(bookmark, state);

            repo.record_op(&[op_ref], &format!("oxy clone {url}"), view)?;

            repo.checkout(&heads, false, "oxy clone (checkout)")?;

        }

        Ok(repo)

    }

#[derive(Debug, Clone, PartialEq, Eq)]

pub struct PullReport {

    pub heads: Vec<ChangeHash>,

    pub changed: bool,

    pub checked_out: bool,

}

        assert_eq!(repo.verify_change(&h2).unwrap(), SignatureStatus::Valid { key, name: "Ada".into(), email: "[email protected]".into(), sealed: false });

    fn clone_push_pull_with_seals_and_attestations() {

        let dir = tempfile::tempdir().unwrap();

        let root = dir.path();

        let origin_path = root.join("origin");

        let origin = Repo::init(&origin_path).unwrap();

        let origin_url = origin_path.to_string_lossy().to_string();

        // A clones, creates an identity (default core.sign = seal), commits, pushes.

        let mut a = Repo::clone(&origin_url, &root.join("a")).unwrap().with_keys_dir(root.join("keys-a"));

        assert_eq!(a.id().unwrap(), origin.id().unwrap());

        let (key_a, _) = a.create_identity("Ada", "[email protected]").unwrap();

        write(&root.join("a"), "f.txt", b"a\n");

        let h1 = a.commit("first", &[]).unwrap();

        let results = a.push("origin", false).unwrap();

        assert!(results[0].ok, "{}", results[0].reason);

        assert_eq!(origin.bookmark_heads().unwrap(), vec![h1]);

        // The change was sealed at push and the reference attested.

        assert!(matches!(origin.authenticate_change(&h1).unwrap(), SignatureStatus::Valid { key, sealed: true, .. } if key == key_a));

        assert_eq!(origin.verify_change(&h1).unwrap(), SignatureStatus::Unsigned);

        let (_, att) = origin.latest_ref_attest("main").unwrap().unwrap();

        assert!(matches!(att.attest.body, AttestBody::Ref { serial: 1, .. }));

        let again = a.push("origin", false).unwrap();

        assert!(again[0].ok && again[0].reason.contains("up to date"));

        let (_, att) = origin.latest_ref_attest("main").unwrap().unwrap();

        assert!(matches!(att.attest.body, AttestBody::Ref { serial: 1, .. }), "a no-op push does not attest again");

        // B clones and sees the file and the seal.

        let b = Repo::clone(&origin_url, &root.join("b")).unwrap();

        assert_eq!(fs::read(root.join("b/f.txt")).unwrap(), b"a\n");

        assert_eq!(b.bookmark_heads().unwrap(), vec![h1]);

        assert!(matches!(b.authenticate_change(&h1).unwrap(), SignatureStatus::Valid { sealed: true, .. }));

        // B (no identity) commits and pushes unsigned; the unverified origin accepts.

        write(&root.join("b"), "f.txt", b"a\nb\n");

        let h2 = b.commit("second", &[]).unwrap();

        assert!(b.push("origin", false).unwrap()[0].ok);

        assert_eq!(origin.bookmark_heads().unwrap(), vec![h2]);

        // A commits an independent change; push is refused until A pulls.

        write(&root.join("a"), "g.txt", b"g\n");

        let h3 = a.commit("third", &[]).unwrap();

        assert!(a.push("origin", false).is_err());

        let pr = a.pull("origin").unwrap();

        assert!(pr.changed && pr.checked_out);

        let mut want = vec![h2, h3];

        want.sort();

        assert_eq!(pr.heads, want);

        assert_eq!(fs::read(root.join("a/f.txt")).unwrap(), b"a\nb\n");

        assert_eq!(fs::read(root.join("a/g.txt")).unwrap(), b"g\n");

        let results = a.push("origin", false).unwrap();

        assert!(results[0].ok, "{}", results[0].reason);

        let mut origin_heads = origin.bookmark_heads().unwrap();

        origin_heads.sort();

        assert_eq!(origin_heads, want);

        let (_, att) = origin.latest_ref_attest("main").unwrap().unwrap();

        assert!(matches!(att.attest.body, AttestBody::Ref { serial: 2, .. }));

        // B pulls and ends up identical.

        let pr = b.pull("origin").unwrap();

        assert_eq!(pr.heads, want);

        assert_eq!(fs::read(root.join("b/g.txt")).unwrap(), b"g\n");

        assert!(!b.snapshot("status").unwrap().changed);

        // Concurrent edits of the same line: the union is a reified conflict.

        write(&root.join("a"), "f.txt", b"a\nA\n");

        a.commit("a-edit", &[]).unwrap();

        a.push("origin", false).unwrap();

        write(&root.join("b"), "f.txt", b"a\nB\n");

        b.commit("b-edit", &[]).unwrap();

        let pr = b.pull("origin").unwrap();

        assert!(pr.changed && pr.checked_out);

        let text = fs::read_to_string(root.join("b/f.txt")).unwrap();

        assert!(text.contains("<<<<<<< oxy conflict"), "{text}");

        write(&root.join("b"), "f.txt", b"a\nAB\n");

        let hr = b.commit("resolve", &[]).unwrap();

        assert_eq!(b.store().get_change(&hr).unwrap().kind, ChangeKind::Resolution);

        assert!(b.push("origin", false).unwrap()[0].ok);

        a.pull("origin").unwrap();

        assert_eq!(fs::read(root.join("a/f.txt")).unwrap(), b"a\nAB\n");

        assert_eq!(a.bookmark_heads().unwrap(), vec![hr]);

    }

    #[test]

    fn receive_push_rejects_bad_updates() {

        let dir = tempfile::tempdir().unwrap();

        let root = dir.path();

        let origin = Repo::init(&root.join("origin")).unwrap();

        let url = root.join("origin").to_string_lossy().to_string();

        let mut a = Repo::clone(&url, &root.join("a")).unwrap().with_keys_dir(root.join("keys"));

        let (key, _) = a.create_identity("Ada", "a@b", ).unwrap();

        write(&root.join("a"), "f.txt", b"x\n");

        let h1 = a.commit("one", &[]).unwrap();

        assert!(a.push("origin", false).unwrap()[0].ok);

        let empty = origin.store().put_state(&StateObject::new(vec![])).unwrap();

        let s1 = origin.store().put_state(&StateObject::new(vec![h1])).unwrap();

        // Missing head.

        let ghost = origin.store().put_state(&StateObject::new(vec![ObjectRef([9; 32])])).unwrap();

        let r = origin.receive_push(Bundle::default(), vec![RefUpdate { name: "main".into(), old_state: s1, new_state: ghost, force: false, attestation: None }]).unwrap();

        assert!(!r[0].ok && r[0].reason.contains("missing"), "{}", r[0].reason);

        // Stale old_state.

        let r = origin.receive_push(Bundle::default(), vec![RefUpdate { name: "main".into(), old_state: empty, new_state: s1, force: false, attestation: None }]).unwrap();

        assert!(!r[0].ok && r[0].reason.contains("stale"), "{}", r[0].reason);

        // Non-fast-forward without force is refused; with force it passes.

        let r = origin.receive_push(Bundle::default(), vec![RefUpdate { name: "main".into(), old_state: s1, new_state: empty, force: false, attestation: None }]).unwrap();

        assert!(!r[0].ok && r[0].reason.contains("fast-forward"), "{}", r[0].reason);

        let r = origin.receive_push(Bundle::default(), vec![RefUpdate { name: "main".into(), old_state: s1, new_state: empty, force: true, attestation: None }]).unwrap();

        assert!(r[0].ok);

        assert_eq!(origin.bookmark_heads().unwrap(), vec![]);

        // Attestation with a stale serial is a rollback and is refused.

        let sa = a.sign_attest(&Attest { repo: a.id().unwrap(), hlc: Hlc::now(), body: AttestBody::Ref { name: "main".into(), old_state: empty, new_state: s1, serial: 1 } }).unwrap();

        let r = origin.receive_push(Bundle::default(), vec![RefUpdate { name: "main".into(), old_state: empty, new_state: s1, force: false, attestation: Some(sa.clone()) }]).unwrap();

        assert!(!r[0].ok && r[0].reason.contains("serial"), "{}", r[0].reason);

        // An attestation for a different update is refused too.

        let sa2 = a.sign_attest(&Attest { repo: a.id().unwrap(), hlc: Hlc::now(), body: AttestBody::Ref { name: "main".into(), old_state: s1, new_state: empty, serial: 5 } }).unwrap();

        let r = origin.receive_push(Bundle::default(), vec![RefUpdate { name: "main".into(), old_state: empty, new_state: s1, force: false, attestation: Some(sa2) }]).unwrap();

        assert!(!r[0].ok && r[0].reason.contains("describe"), "{}", r[0].reason);

        let _ = key;

    }

    #[test]

use crate::attest::Attest;

    pub fn put_attest(&self, a: &Attest) -> Result<ObjectRef> {

        self.put(ObjectType::Attest, &a.encode())

    }

    pub fn get_attest(&self, r: &ObjectRef) -> Result<Attest> {

        Ok(Attest::decode(&self.get_typed(r, ObjectType::Attest)?)?)

    }

    fn seal_index_path(&self, change: &ObjectRef) -> PathBuf {

        let h = change.to_hex();

        self.dir.join("seals").join(&h[..2]).join(&h[2..])

    }

    /// Record that seal attestation `attest` covers `change`.

    pub fn index_seal(&self, change: &ObjectRef, attest: &ObjectRef) -> Result<()> {

        let mut seals = self.seals_for(change)?;

        if seals.contains(attest) {

            return Ok(());

        }

        seals.push(*attest);

        seals.sort();

        let path = self.seal_index_path(change);

        fs::create_dir_all(path.parent().unwrap())?;

        let text: String = seals.iter().map(|s| format!("{}\n", s.to_hex())).collect();

        let tmp = path.with_extension("tmp");

        fs::write(&tmp, text)?;

        fs::rename(&tmp, &path)?;

        Ok(())

    }

    /// Seal attestations known to cover `change`.

    pub fn seals_for(&self, change: &ObjectRef) -> Result<Vec<ObjectRef>> {

        match fs::read_to_string(self.seal_index_path(change)) {

            Ok(t) => Ok(t.lines().filter_map(ObjectRef::from_hex).collect()),

            Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(vec![]),

            Err(e) => Err(e.into()),

        }

    }

    /// Clone a repository from a local path

    Clone { url: String, dest: Option<PathBuf> },

    /// Manage remotes

    Remote {

        #[command(subcommand)]

        cmd: RemoteCmd,

    },

    /// Fetch every bookmark of a remote into its tracking references

    Fetch { remote: Option<String> },

    /// Fetch, then union the remote bookmark into the local one (no merge change)

    Pull { remote: Option<String> },

    /// Push the current bookmark to a remote

    Push {

        remote: Option<String>,

        /// Allow dropping heads the remote has

        #[arg(long)]

        force: bool,

    },

enum RemoteCmd {

    Add { name: String, url: String },

    Remove { name: String },

    List,

}

#[derive(Subcommand)]

    Ok(match repo.authenticate_change(h)? {

        SignatureStatus::Valid { key, name, email, sealed } => {

            format!("{} by {name} <{email}> key {}", if sealed { "sealed" } else { "signed" }, key.short(12))

        }

        Cmd::Clone { url, dest } => {

            let dest = dest.unwrap_or_else(|| {

                let base = url.trim_end_matches('/').rsplit('/').next().unwrap_or("repo");

                PathBuf::from(base)

            });

            let repo = Repo::clone(&url, &dest)?;

            let heads = repo.bookmark_heads()?;

            println!("Cloned into {} ({} on {}: {} head{})", repo.root().display(), url, repo.config.bookmark, heads.len(), if heads.len() == 1 { "" } else { "s" });

            Ok(0)

        }

        Cmd::Remote { cmd } => {

            let mut repo = open()?;

            match cmd {