refuse misplaced authors

5 files changed, 267 insertions, 8 deletions

diff --git a/src/document/changes.rs b/src/document/changes.rs
index 7ee1a4b..57e2bf7 100644
--- a/src/document/changes.rs
+++ b/src/document/changes.rs
@@ -158,11 +158,127 @@ pub(crate) fn delta_bytes(
 
 /// Loads a delta (`delta_bytes`) into `doc`.
 pub(crate) fn load_delta(doc: &mut Document, bytes: &[u8]) -> Result<()> {
+    if let Some(misplaced) = misplaced_author(bytes) {
+        bail!("Refused the downloaded changes: {misplaced}");
+    }
     doc.load_incremental(bytes)
         .context("Loading the downloaded changes")?;
     Ok(())
 }
 
+/// A change that names an author although it isn't its actor's first (`misplaced_author`).
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) struct MisplacedAuthor {
+    pub(crate) author: Vec<u8>,
+    pub(crate) actor: super::ActorId,
+    pub(crate) seq: u64,
+}
+
+impl std::fmt::Display for MisplacedAuthor {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "change {} of actor {} names an author ({}), which only an actor's first change may",
+            self.seq,
+            self.actor,
+            String::from_utf8_lossy(&self.author)
+        )
+    }
+}
+
+/// The first change among those in `data` (chunks as `load_incremental` takes them:
+/// changes, compressed changes, bundles, documents) that names an author although it
+/// isn't its actor's first change. Automerge records an author on an actor's first change
+/// only, and applying a later one that carries an author trips an assertion in Automerge
+/// 0.12 (`ChangeGraph::add_changes`) halfway through updating the document: the task
+/// panics, and the document is left inconsistent. No honest node makes such a change, so
+/// whatever comes from elsewhere is looked at here first, until Automerge refuses such
+/// changes itself. Data that doesn't parse is left to Automerge, which refuses it.
+///
+/// A document chunk is loaded on its own to read its changes (no assertion on that
+/// path): about as much work again as applying it, paid for documents in sync messages
+/// (Automerge sends one when a peer lacks a third of the history or more).
+pub(crate) fn misplaced_author(mut data: &[u8]) -> Option<MisplacedAuthor> {
+    const MAGIC: [u8; 4] = [0x85, 0x6f, 0x4a, 0x83];
+    // Magic bytes, checksum, chunk type, then the body's length (LEB128) and the body.
+    const HEADER: usize = 9;
+    while data.len() > HEADER && data[..4] == MAGIC {
+        let kind = data[8];
+        let (length, read) = uleb128(&data[HEADER..])?;
+        let end = HEADER
+            .checked_add(read)?
+            .checked_add(usize::try_from(length).ok()?)?;
+        let chunk = data.get(..end)?;
+        data = &data[end..];
+        let found = match kind {
+            // A change, or a compressed one.
+            1 | 2 => Change::try_from(chunk)
+                .ok()
+                .and_then(|change| misplaced_in(&change)),
+            3 => automerge::Bundle::try_from(chunk)
+                .ok()
+                .and_then(|bundle| bundle.to_changes().ok())
+                .and_then(|changes| changes.iter().find_map(misplaced_in)),
+            0 => Document::load(chunk).ok().and_then(|doc| {
+                doc.get_changes_meta(&[]).into_iter().find_map(|meta| {
+                    (meta.seq != 1)
+                        .then(|| author_footer(&meta.extra))
+                        .flatten()
+                        .map(|author| MisplacedAuthor {
+                            author,
+                            actor: meta.actor.into_owned(),
+                            seq: meta.seq,
+                        })
+                })
+            }),
+            _ => None,
+        };
+        if found.is_some() {
+            return found;
+        }
+    }
+    None
+}
+
+/// `change`, if it names an author but isn't its actor's first (`misplaced_author`).
+fn misplaced_in(change: &Change) -> Option<MisplacedAuthor> {
+    let author = change.author()?;
+    (change.seq() != 1).then(|| MisplacedAuthor {
+        author: author.as_bytes().to_vec(),
+        actor: change.actor_id().clone(),
+        seq: change.seq(),
+    })
+}
+
+/// The author in a change's extra bytes: footers, each its kind (LEB128; 1 is the
+/// author), its length (LEB128) and its bytes.
+fn author_footer(mut extra: &[u8]) -> Option<Vec<u8>> {
+    while !extra.is_empty() {
+        let (kind, read) = uleb128(extra)?;
+        let (length, more) = uleb128(&extra[read..])?;
+        let start = read + more;
+        let end = start.checked_add(usize::try_from(length).ok()?)?;
+        let bytes = extra.get(start..end)?;
+        if kind == 1 {
+            return Some(bytes.to_vec());
+        }
+        extra = &extra[end..];
+    }
+    None
+}
+
+/// An unsigned LEB128 number at the start of `bytes`, and how many bytes it took.
+fn uleb128(bytes: &[u8]) -> Option<(u64, usize)> {
+    let mut value = 0u64;
+    for (i, byte) in bytes.iter().enumerate().take(10) {
+        value |= u64::from(byte & 0x7f) << (7 * i);
+        if byte & 0x80 == 0 {
+            return Some((value, i + 1));
+        }
+    }
+    None
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -307,4 +423,55 @@ mod tests {
         let (_, unknown) = authors_since(&snapshot, &merged.get_heads());
         assert!(unknown);
     }
+
+    /// An actor without an author, and its two changes: the second as made, and with an
+    /// author footer it mustn't carry.
+    fn misplaced() -> (Vec<u8>, Vec<u8>, Vec<u8>) {
+        let mut doc = Document::new().with_actor(ActorId::from(b"writer".to_vec()));
+        change(&mut doc, |tx| tx.put(ROOT, "x", 1).unwrap());
+        change(&mut doc, |tx| tx.put(ROOT, "x", 2).unwrap());
+        let changes = doc.get_changes(&[]);
+        let (first, second) = (
+            changes[0].raw_bytes().to_vec(),
+            changes[1].raw_bytes().to_vec(),
+        );
+        let crafted = crate::document::testing::with_author_footer(&second, b"mallory");
+        (first, second, crafted)
+    }
+
+    #[test]
+    fn an_author_on_a_later_change_is_found() {
+        let (first, second, crafted) = misplaced();
+        assert_eq!(misplaced_author(&[first.clone(), second].concat()), None);
+        let found = misplaced_author(&[first.clone(), crafted.clone()].concat()).unwrap();
+        assert_eq!(found.seq, 2);
+        assert_eq!(found.author, b"mallory");
+        assert_eq!(found.actor, ActorId::from(b"writer".to_vec()));
+        // Honest authors (on first changes) and whole documents pass.
+        let mut authored = Document::new();
+        authored.set_author(Some(automerge::Author::from(b"node".to_vec())));
+        change(&mut authored, |tx| tx.put(ROOT, "a", 1).unwrap());
+        change(&mut authored, |tx| tx.put(ROOT, "a", 2).unwrap());
+        assert_eq!(misplaced_author(&authored.save()), None);
+        let bytes: Vec<u8> = authored
+            .get_changes(&[])
+            .iter()
+            .flat_map(|change| change.raw_bytes().to_vec())
+            .collect();
+        assert_eq!(misplaced_author(&bytes), None);
+        // What isn't chunks is left to Automerge.
+        assert_eq!(misplaced_author(b"not a chunk"), None);
+    }
+
+    #[test]
+    fn a_delta_with_a_misplaced_author_is_refused_and_the_document_kept() {
+        let (first, _, crafted) = misplaced();
+        let mut doc = Document::new();
+        load_delta(&mut doc, &first).unwrap();
+        let heads = doc.get_heads();
+        assert!(load_delta(&mut doc, &crafted).is_err());
+        assert_eq!(doc.get_heads(), heads);
+        // Still whole: it saves and loads.
+        crate::document::load(&doc.save()).unwrap();
+    }
 }
diff --git a/src/document/sync.rs b/src/document/sync.rs
index dab43b4..8cb740b 100644
--- a/src/document/sync.rs
+++ b/src/document/sync.rs
@@ -10,7 +10,6 @@ use std::collections::HashSet;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::time::Duration;
 
-use automerge::AutomergeError;
 use automerge::sync::{self, SyncDoc};
 use tokio::time::Instant;
 use tracing::debug;
@@ -302,23 +301,29 @@ impl SyncSession {
     /// Applies a sync message from the peer to `doc`. Doesn't settle the awaited
     /// answer: the caller measures the exchange first (`reply`).
     #[cfg(test)]
-    pub(crate) fn receive(
-        &mut self,
-        doc: &mut Document,
-        message: Received,
-    ) -> Result<(), AutomergeError> {
+    pub(crate) fn receive(&mut self, doc: &mut Document, message: Received) -> anyhow::Result<()> {
         self.receive_logged(doc, message, &mut automerge::PatchLog::inactive())
     }
 
     /// Applies a sync message from the peer to `doc`, what it changes recorded in `log`
     /// (`Replica::logged`). Doesn't settle the awaited answer: the caller measures the
-    /// exchange first (`reply`).
+    /// exchange first (`reply`). A message carrying a change that names an author though
+    /// it isn't its actor's first is refused whole, before Automerge sees it
+    /// (`changes::misplaced_author`).
     pub(crate) fn receive_logged(
         &mut self,
         doc: &mut Document,
         message: Received,
         log: &mut automerge::PatchLog,
-    ) -> Result<(), AutomergeError> {
+    ) -> anyhow::Result<()> {
+        if let Some(misplaced) = message
+            .0
+            .changes
+            .iter()
+            .find_map(crate::document::changes::misplaced_author)
+        {
+            anyhow::bail!("Refused a sync message: {misplaced}");
+        }
         self.delta_offered = false;
         self.before_receive = Some(self.peer_view());
         let said = message.0.heads.clone();
@@ -1909,6 +1914,34 @@ mod tests {
         assert_eq!(theirs.get_heads(), ours.get_heads(), "both in one message");
     }
 
+    /// A change naming an author though it isn't its actor's first is refused before
+    /// Automerge sees it (`changes::misplaced_author`), and the document stays as it was.
+    #[test]
+    fn a_message_with_a_misplaced_author_is_refused() {
+        use crate::document::testing::{change, with_author_footer};
+        use automerge::{ActorId, ROOT, transaction::Transactable};
+        let mut writer = Document::new().with_actor(ActorId::from(b"writer".to_vec()));
+        change(&mut writer, |doc| doc.put(ROOT, "x", 1).unwrap());
+        change(&mut writer, |doc| doc.put(ROOT, "x", 2).unwrap());
+        let changes = writer.get_changes(&[]);
+        let first = changes[0].raw_bytes().to_vec();
+        let crafted = with_author_footer(changes[1].raw_bytes(), b"mallory");
+        let (mut ours, mut theirs) = (Document::new(), Document::new());
+        let (mut a, _) = in_step(&mut ours, &mut theirs);
+        a.receive(
+            &mut ours,
+            Received::pushed(Vec::new(), vec![first]).unwrap(),
+        )
+        .unwrap();
+        let heads = ours.get_heads();
+        let refused = a.receive(
+            &mut ours,
+            Received::pushed(Vec::new(), vec![crafted]).unwrap(),
+        );
+        assert!(refused.is_err());
+        assert_eq!(ours.get_heads(), heads);
+    }
+
     /// Pushed changes (`push`) do what the sync message they stand for would: the peer
     /// applies them, is level with us, and has nothing to answer; nothing goes twice.
     #[test]
diff --git a/src/document/testing.rs b/src/document/testing.rs
index 7b75a96..cd6c15b 100644
--- a/src/document/testing.rs
+++ b/src/document/testing.rs
@@ -29,3 +29,48 @@ pub(crate) fn try_change<T, E>(
     }
     out
 }
+
+/// `raw` (an uncompressed change chunk, as `Change::raw_bytes` gives one) with an author
+/// footer appended to its extra bytes, its checksum made right again: what no honest
+/// node sends for a change that isn't its actor's first (`changes::misplaced_author`).
+pub(crate) fn with_author_footer(raw: &[u8], author: &[u8]) -> Vec<u8> {
+    use sha2::{Digest, Sha256};
+    fn leb(mut n: usize, out: &mut Vec<u8>) {
+        loop {
+            let byte = (n & 0x7f) as u8;
+            n >>= 7;
+            if n == 0 {
+                out.push(byte);
+                return;
+            }
+            out.push(byte | 0x80);
+        }
+    }
+    assert_eq!(raw[8], 1, "an uncompressed change chunk");
+    let (mut length, mut read, mut shift) = (0usize, 0, 0);
+    loop {
+        let byte = raw[9 + read];
+        length |= usize::from(byte & 0x7f) << shift;
+        read += 1;
+        if byte & 0x80 == 0 {
+            break;
+        }
+        shift += 7;
+    }
+    let mut body = raw[9 + read..9 + read + length].to_vec();
+    body.push(1);
+    leb(author.len(), &mut body);
+    body.extend_from_slice(author);
+    let mut header = Vec::new();
+    leb(body.len(), &mut header);
+    let mut summed = vec![1u8];
+    summed.extend(&header);
+    summed.extend(&body);
+    let sum = Sha256::digest(&summed);
+    let mut out = raw[..4].to_vec();
+    out.extend(&sum[..4]);
+    out.push(1);
+    out.extend(header);
+    out.extend(body);
+    out
+}
diff --git a/src/remake/kind.rs b/src/remake/kind.rs
index 03f2ef7..a8d80d8 100644
--- a/src/remake/kind.rs
+++ b/src/remake/kind.rs
@@ -374,6 +374,9 @@ fn load_encoded(encoded: &str) -> Result<Document, String> {
     let bytes = BASE64
         .decode(encoded.trim())
         .map_err(|error| format!("the document is not valid base64: {error}"))?;
+    if let Some(misplaced) = crate::document::changes::misplaced_author(&bytes) {
+        return Err(format!("the document can't be loaded: {misplaced}"));
+    }
     Document::load(&bytes)
         .map_err(|error| format!("the document is not an Automerge document: {error}"))
 }
@@ -410,6 +413,12 @@ impl RemakeKind for Import {
         target: &mut Transaction<'_>,
     ) -> Result<(), StoreError> {
         let content = content.ok_or_else(|| anyhow::anyhow!("import without content"))?;
+        // From the coordinator: looked at before Automerge loads it (`misplaced_author`).
+        if let Some(misplaced) = crate::document::changes::misplaced_author(content) {
+            return Err(
+                anyhow::anyhow!("the imported document can't be loaded: {misplaced}").into(),
+            );
+        }
         let imported = Document::load(content)?;
         // Deterministic: the same bytes give the same view, copied in a fixed order.
         store::materialize_into(&imported, target, &meta::is_protocol)
diff --git a/src/state/host.rs b/src/state/host.rs
index 5751998..b4a785b 100644
--- a/src/state/host.rs
+++ b/src/state/host.rs
@@ -374,6 +374,11 @@ impl Host for AppState {
     ) -> impl Future<Output = Result<(), String>> + Send {
         let state = self.clone();
         async move {
+            // Downloaded: looked at before Automerge applies it (`misplaced_author`).
+            if let Some(misplaced) = crate::document::changes::misplaced_author(change.raw_bytes())
+            {
+                return Err(misplaced.to_string());
+            }
             tokio::task::spawn_blocking(move || {
                 let result = {
                     let mut replica = state.doc.blocking_write();