openpgp: Make serialize::stream::Message Send + Sync.

  - See #615.

18 files changed, 164 insertions, 89 deletions

diff --git a/autocrypt/src/lib.rs b/autocrypt/src/lib.rs
index a82a1952..3b52cb69 100644
--- a/autocrypt/src/lib.rs
+++ b/autocrypt/src/lib.rs
@@ -480,7 +480,9 @@ impl AutocryptSetupMessage {
     /// Generates the Autocrypt Setup Message.
     ///
     /// The message is written to `w`.
-    pub fn serialize<W: io::Write>(&mut self, w: &mut W) -> Result<()> {
+    pub fn serialize<W>(&mut self, w: &mut W) -> Result<()>
+        where W: io::Write + Send + Sync,
+    {
         // The outer message is an ASCII-armored encoded message
         // containing a single SK-ESK and a single SEIP packet.  The
         // SEIP packet contains a literal data packet whose content is
diff --git a/guide/src/chapter_01.md b/guide/src/chapter_01.md
index 874b70e6..0a60b1ea 100644
--- a/guide/src/chapter_01.md
+++ b/guide/src/chapter_01.md
@@ -55,13 +55,13 @@ fn main() -> openpgp::Result<()> {
 #
 # /// Signs the given message.
 # fn sign(policy: &dyn Policy,
-#         sink: &mut dyn Write, plaintext: &str, tsk: &openpgp::Cert)
+#         sink: &mut (dyn Write + Send + Sync), plaintext: &str, tsk: &openpgp::Cert)
 #     -> openpgp::Result<()>
 # {
 #     // Get the keypair to do the signing from the Cert.
 #     let keypair = tsk
 #         .keys().unencrypted_secret()
-#         .with_policy(policy, None).supported().alive().revoked(false).for_signing()
+#         .with_policy(policy, None).alive().revoked(false).for_signing()
 #         .nth(0).unwrap().key().clone().into_keypair()?;
 #
 #     // Start streaming an OpenPGP message.
@@ -206,13 +206,13 @@ fn generate() -> openpgp::Result<openpgp::Cert> {
 #
 # /// Signs the given message.
 # fn sign(policy: &dyn Policy,
-#         sink: &mut dyn Write, plaintext: &str, tsk: &openpgp::Cert)
+#         sink: &mut (dyn Write + Send + Sync), plaintext: &str, tsk: &openpgp::Cert)
 #     -> openpgp::Result<()>
 # {
 #     // Get the keypair to do the signing from the Cert.
 #     let keypair = tsk
 #         .keys().unencrypted_secret()
-#         .with_policy(policy, None).supported().alive().revoked(false).for_signing()
+#         .with_policy(policy, None).alive().revoked(false).for_signing()
 #         .nth(0).unwrap().key().clone().into_keypair()?;
 #
 #     // Start streaming an OpenPGP message.
@@ -357,13 +357,13 @@ implements [`io::Write`], and we simply write the plaintext to it.
 #
 /// Signs the given message.
 fn sign(policy: &dyn Policy,
-        sink: &mut dyn Write, plaintext: &str, tsk: &openpgp::Cert)
+        sink: &mut (dyn Write + Send + Sync), plaintext: &str, tsk: &openpgp::Cert)
     -> openpgp::Result<()>
 {
     // Get the keypair to do the signing from the Cert.
     let keypair = tsk
         .keys().unencrypted_secret()
-        .with_policy(policy, None).supported().alive().revoked(false).for_signing()
+        .with_policy(policy, None).alive().revoked(false).for_signing()
         .nth(0).unwrap().key().clone().into_keypair()?;
 
     // Start streaming an OpenPGP message.
@@ -519,13 +519,13 @@ Verified data can be read from this using [`io::Read`].
 #
 # /// Signs the given message.
 # fn sign(policy: &dyn Policy,
-#         sink: &mut dyn Write, plaintext: &str, tsk: &openpgp::Cert)
+#         sink: &mut (dyn Write + Send + Sync), plaintext: &str, tsk: &openpgp::Cert)
 #     -> openpgp::Result<()>
 # {
 #     // Get the keypair to do the signing from the Cert.
 #     let keypair = tsk
 #         .keys().unencrypted_secret()
-#         .with_policy(policy, None).supported().alive().revoked(false).for_signing()
+#         .with_policy(policy, None).alive().revoked(false).for_signing()
 #         .nth(0).unwrap().key().clone().into_keypair()?;
 #
 #     // Start streaming an OpenPGP message.
diff --git a/guide/src/chapter_02.md b/guide/src/chapter_02.md
index 7d663701..fc88f653 100644
--- a/guide/src/chapter_02.md
+++ b/guide/src/chapter_02.md
@@ -55,7 +55,7 @@ fn main() -> openpgp::Result<()> {
 #
 # /// Encrypts the given message.
 # fn encrypt(policy: &dyn Policy,
-#            sink: &mut dyn Write, plaintext: &str, recipient: &openpgp::Cert)
+#            sink: &mut (dyn Write + Send + Sync), plaintext: &str, recipient: &openpgp::Cert)
 #            -> openpgp::Result<()> {
 #    let recipients =
 #        recipient.keys().with_policy(policy, None).supported().alive().revoked(false)
@@ -203,7 +203,7 @@ fn generate() -> openpgp::Result<openpgp::Cert> {
 #
 # /// Encrypts the given message.
 # fn encrypt(policy: &dyn Policy,
-#            sink: &mut dyn Write, plaintext: &str, recipient: &openpgp::Cert)
+#            sink: &mut (dyn Write + Send + Sync), plaintext: &str, recipient: &openpgp::Cert)
 #            -> openpgp::Result<()> {
 #    let recipients =
 #        recipient.keys().with_policy(policy, None).supported().alive().revoked(false)
@@ -351,7 +351,7 @@ implements [`io::Write`], and we simply write the plaintext to it.
 #
 /// Encrypts the given message.
 fn encrypt(policy: &dyn Policy,
-           sink: &mut dyn Write, plaintext: &str, recipient: &openpgp::Cert)
+           sink: &mut (dyn Write + Send + Sync), plaintext: &str, recipient: &openpgp::Cert)
            -> openpgp::Result<()> {
     let recipients =
         recipient.keys().with_policy(policy, None).supported().alive().revoked(false)
@@ -513,7 +513,7 @@ Decrypted data can be read from this using [`io::Read`].
 #
 # /// Encrypts the given message.
 # fn encrypt(policy: &dyn Policy,
-#            sink: &mut dyn Write, plaintext: &str, recipient: &openpgp::Cert)
+#            sink: &mut (dyn Write + Send + Sync), plaintext: &str, recipient: &openpgp::Cert)
 #            -> openpgp::Result<()> {
 #    let recipients =
 #        recipient.keys().with_policy(policy, None).supported().alive().revoked(false)
diff --git a/openpgp-ffi/include/sequoia/io.h b/openpgp-ffi/include/sequoia/io.h
index 81d0093a..e574aefb 100644
--- a/openpgp-ffi/include/sequoia/io.h
+++ b/openpgp-ffi/include/sequoia/io.h
@@ -92,6 +92,13 @@ pgp_writer_t pgp_writer_from_bytes (uint8_t *buf, size_t len);
 /// reference a chunk of memory allocated using libc's heap allocator.
 /// The caller is responsible to `free` it once the writer has been
 /// destroyed.
+///
+/// # Sending objects across thread boundaries
+///
+/// If you send a Sequoia object (like a pgp_writer_stack_t) that
+/// serializes to an allocating writer across thread boundaries, you
+/// must make sure that the system's allocator (i.e. `realloc (3)`)
+/// supports reallocating memory allocated by another thread.
 /*/
 pgp_writer_t pgp_writer_alloc (void **buf, size_t *len);
 
@@ -104,6 +111,12 @@ typedef ssize_t (*pgp_writer_cb_t) (void *cookie, const void *buf, size_t len);
 /// Creates an writer from a callback and cookie.
 ///
 /// This writer calls the given callback to write data.
+///
+/// # Sending objects across thread boundaries
+///
+/// If you send a Sequoia object (like a pgp_writer_stack_t) that
+/// serializes to a callback-based writer across thread boundaries,
+/// you must make sure that the callback and cookie also support this.
 /*/
 pgp_writer_t pgp_writer_from_callback (pgp_writer_cb_t, void *);
 
diff --git a/openpgp-ffi/src/io.rs b/openpgp-ffi/src/io.rs
index 35e80c47..484615cf 100644
--- a/openpgp-ffi/src/io.rs
+++ b/openpgp-ffi/src/io.rs
@@ -4,6 +4,7 @@ use std::fs::File;
 use std::io::{self, Read, Write, Cursor};
 use std::path::Path;
 use std::slice;
+use std::sync::Mutex;
 use libc::{c_void, c_char, size_t, ssize_t, realloc};
 
 #[cfg(unix)]
@@ -181,7 +182,7 @@ pub struct Writer(WriterKind);
 /// In some cases, we want to call functions on concrete types.  To
 /// avoid nasty hacks, we have specialized variants for that.
 pub(crate) enum WriterKind {
-    Generic(Box<dyn io::Write>),
+    Generic(Box<dyn io::Write + Send + Sync>),
     Armored(openpgp::armor::Writer<&'static mut WriterKind>),
 }
 
@@ -247,6 +248,13 @@ fn pgp_writer_from_bytes(buf: *mut u8, len: size_t) -> *mut Writer {
 /// The caller is responsible to `free` it once the writer has been
 /// destroyed.
 ///
+/// # Sending objects across thread boundaries
+///
+/// If you send a Sequoia object (like a pgp_writer_stack_t) that
+/// serializes to an allocating writer across thread boundaries, you
+/// must make sure that the system's allocator (i.e. `realloc (3)`)
+/// supports reallocating memory allocated by another thread.
+///
 /// # Examples
 ///
 /// ```c
@@ -302,32 +310,36 @@ fn pgp_writer_alloc(buf: *mut *mut c_void, len: *mut size_t)
     let buf = ffi_param_ref_mut!(buf);
     let len = ffi_param_ref_mut!(len);
 
-    let w = WriterKind::Generic(Box::new(WriterAlloc {
+    let w = WriterKind::Generic(Box::new(WriterAlloc(Mutex::new(Buffer {
         buf: buf,
         len: len,
-    }));
+    }))));
     w.move_into_raw()
 }
 
-struct WriterAlloc {
+struct WriterAlloc(Mutex<Buffer>);
+struct Buffer {
     buf: &'static mut *mut c_void,
     len: &'static mut size_t,
 }
 
+unsafe impl Send for Buffer {}
+
 impl Write for WriterAlloc {
     fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
-        let old_len = *self.len;
+        let buffer = self.0.get_mut().expect("Mutex not to be poisoned");
+        let old_len = *buffer.len;
         let new_len = old_len + buf.len();
 
         let new = unsafe {
-            realloc(*self.buf, new_len)
+            realloc(*buffer.buf, new_len)
         };
         if new.is_null() {
             return Err(io::Error::new(io::ErrorKind::Other, "out of memory"));
         }
 
-        *self.buf = new;
-        *self.len = new_len;
+        *buffer.buf = new;
+        *buffer.len = new_len;
 
         let sl = unsafe {
             slice::from_raw_parts_mut(new as *mut u8, new_len)
@@ -348,26 +360,37 @@ type WriterCallbackFn = extern fn(*mut c_void, *const c_void, size_t) -> ssize_t
 /// Creates an writer from a callback and cookie.
 ///
 /// This writer calls the given callback to write data.
+///
+/// # Sending objects across thread boundaries
+///
+/// If you send a Sequoia object (like a pgp_writer_stack_t) that
+/// serializes to a callback-based writer across thread boundaries,
+/// you must make sure that the callback and cookie also support this.
 #[::sequoia_ffi_macros::extern_fn] #[no_mangle] pub extern "C"
 fn pgp_writer_from_callback(cb: WriterCallbackFn,
                             cookie: *mut c_void)
                             -> *mut Writer {
-    let w = WriterKind::Generic(Box::new(WriterCallback {
-        cb, cookie,
-    }));
+    let w = WriterKind::Generic(Box::new(WriterCallback(Mutex::new(
+        WriterClosure { cb, cookie, }))));
     w.move_into_raw()
 }
 
 /// A generic callback-based writer implementation.
-struct WriterCallback {
+struct WriterCallback(Mutex<WriterClosure>);
+
+struct WriterClosure {
     cb: WriterCallbackFn,
     cookie: *mut c_void,
 }
 
+unsafe impl Send for WriterClosure {}
+
 impl Write for WriterCallback {
     fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        let closure = self.0.get_mut().expect("Mutex not to be poisoned");
         let r =
-            (self.cb)(self.cookie, buf.as_ptr() as *const c_void, buf.len());
+            (closure.cb)(closure.cookie,
+                         buf.as_ptr() as *const c_void, buf.len());
         if r < 0 {
             use std::io as stdio;
             Err(stdio::Error::new(stdio::ErrorKind::Other,
diff --git a/openpgp-ffi/src/serialize.rs b/openpgp-ffi/src/serialize.rs
index aecf1a0a..053fedbb 100644
--- a/openpgp-ffi/src/serialize.rs
+++ b/openpgp-ffi/src/serialize.rs
@@ -152,7 +152,7 @@ pub extern "C" fn pgp_arbitrary_writer_new
 pub extern "C" fn pgp_signer_new
     (errp: Option<&mut *mut crate::error::Error>,
      inner: *mut Message<'static>,
-     signers: *const *mut Box<dyn self::openpgp::crypto::Signer>,
+     signers: *const *mut Box<dyn self::openpgp::crypto::Signer + Send + Sync>,
      signers_len: size_t,
      hash_algo: u8)
      -> *mut Message<'static>
@@ -189,7 +189,7 @@ pub extern "C" fn pgp_signer_new
 pub extern "C" fn pgp_signer_new_detached
     (errp: Option<&mut *mut crate::error::Error>,
      inner: *mut Message<'static>,
-     signers: *const *mut Box<dyn self::openpgp::crypto::Signer>,
+     signers: *const *mut Box<dyn self::openpgp::crypto::Signer + Send + Sync>,
      signers_len: size_t,
      hash_algo: u8)
      -> *mut Message<'static>
diff --git a/openpgp/examples/generate-encrypt-decrypt.rs b/openpgp/examples/generate-encrypt-decrypt.rs
index bc147827..821dc63a 100644
--- a/openpgp/examples/generate-encrypt-decrypt.rs
+++ b/openpgp/examples/generate-encrypt-decrypt.rs
@@ -46,8 +46,8 @@ fn generate() -> openpgp::Result<openpgp::Cert> {
 }
 
 /// Encrypts the given message.
-fn encrypt(p: &dyn Policy, sink: &mut dyn Write, plaintext: &str,
-           recipient: &openpgp::Cert)
+fn encrypt(p: &dyn Policy, sink: &mut (dyn Write + Send + Sync),
+           plaintext: &str, recipient: &openpgp::Cert)
     -> openpgp::Result<()>
 {
     let recipients =
diff --git a/openpgp/examples/generate-sign-verify.rs b/openpgp/examples/generate-sign-verify.rs
index ee7b67b2..6242b623 100644
--- a/openpgp/examples/generate-sign-verify.rs
+++ b/openpgp/examples/generate-sign-verify.rs
@@ -44,7 +44,8 @@ fn generate() -> openpgp::Result<openpgp::Cert> {
 }
 
 /// Signs the given message.
-fn sign(p: &dyn Policy, sink: &mut dyn Write, plaintext: &str, tsk: &openpgp::Cert)
+fn sign(p: &dyn Policy, sink: &mut (dyn Write + Send + Sync),
+        plaintext: &str, tsk: &openpgp::Cert)
            -> openpgp::Result<()> {
     // Get the keypair to do the signing from the Cert.
     let keypair = tsk
diff --git a/openpgp/src/crypto/asymmetric.rs b/openpgp/src/crypto/asymmetric.rs
index 1934f3be..89941673 100644
--- a/openpgp/src/crypto/asymmetric.rs
+++ b/openpgp/src/crypto/asymmetric.rs
@@ -56,6 +56,17 @@ impl Signer for Box<dyn Signer> {
     }
 }
 
+impl Signer for Box<dyn Signer + Send + Sync> {
+    fn public(&self) -> &Key<key::PublicParts, key::UnspecifiedRole> {
+        self.as_ref().public()
+    }
+
+    fn sign(&mut self, hash_algo: HashAlgorithm, digest: &[u8])
+            -> Result<mpi::Signature> {
+        self.as_mut().sign(hash_algo, digest)
+    }
+}
+
 /// Decrypts a message.
 ///
 /// Used by [`PKESK::decrypt`] to decrypt session keys.
diff --git a/openpgp/src/serialize.rs b/openpgp/src/serialize.rs
index 93657ae6..5795183b 100644
--- a/openpgp/src/serialize.rs
+++ b/openpgp/src/serialize.rs
@@ -2153,6 +2153,22 @@ impl Marshal for CompressedData {
     /// This function works recursively: if the `CompressedData` packet
     /// contains any packets, they are also serialized.
     fn serialize(&self, o: &mut dyn std::io::Write) -> Result<()> {
+        // The streaming serialization framework requires the sink to
+        // be Send + Sync, but `o` is not.  Knowing that we create the
+        // message here and don't keep the message object around, we
+        // can cheat by creating a shim that is Send + Sync.
+        struct Shim<'a>(&'a mut dyn std::io::Write);
+        impl std::io::Write for Shim<'_> {
+            fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+                self.0.write(buf)
+            }
+            fn flush(&mut self) -> std::io::Result<()> {
+                self.0.flush()
+            }
+        }
+        unsafe impl Send for Shim<'_> {}
+        unsafe impl Sync for Shim<'_> {}
+
         match self.body() {
             Body::Unprocessed(bytes) => {
                 if TRACE {
@@ -2172,7 +2188,7 @@ impl Marshal for CompressedData {
                               self.algo(), bytes.len());
                 }
 
-                let o = stream::Message::new(o);
+                let o = stream::Message::new(Shim(o));
                 let mut o = stream::Compressor::new_naked(
                     o, self.algo(), Default::default(), 0)?;
                 o.write_all(bytes)?;
@@ -2186,7 +2202,7 @@ impl Marshal for CompressedData {
                               self.algo(), children.len());
                 }
 
-                let o = stream::Message::new(o);
+                let o = stream::Message::new(Shim(o));
                 let mut o = stream::Compressor::new_naked(
                     o, self.algo(), Default::default(), 0)?;
 
diff --git a/openpgp/src/serialize/stream.rs b/openpgp/src/serialize/stream.rs
index ed32e4c2..0cb96178 100644
--- a/openpgp/src/serialize/stream.rs
+++ b/openpgp/src/serialize/stream.rs
@@ -226,7 +226,7 @@ impl<'a> Message<'a> {
     /// message.finalize()?;
     /// # Ok(()) }
     /// ```
-    pub fn new<W: 'a + io::Write>(w: W) -> Message<'a> {
+    pub fn new<W: 'a + io::Write + Send + Sync>(w: W) -> Message<'a> {
         writer::Generic::new(w, Cookie::new(0))
     }
 
@@ -322,8 +322,8 @@ impl<'a> Message<'a> {
     }
 }
 
-impl<'a> From<&'a mut dyn io::Write> for Message<'a> {
-    fn from(w: &'a mut dyn io::Write) -> Self {
+impl<'a> From<&'a mut (dyn io::Write + Send + Sync)> for Message<'a> {
+    fn from(w: &'a mut (dyn io::Write + Send + Sync)) -> Self {
         writer::Generic::new(w, Cookie::new(0))
     }
 }
@@ -597,10 +597,10 @@ impl<'a> writer::Stackable<'a, Cookie> for ArbitraryWriter<'a> {
     fn mount(&mut self, _new: writer::BoxStack<'a, Cookie>) {
         unreachable!("Only implemented by Signer")
     }
-    fn inner_ref(&self) -> Option<&dyn writer::Stackable<'a, Cookie>> {
+    fn inner_ref(&self) -> Option<&(dyn writer::Stackable<'a, Cookie> + Send + Sync)> {
         Some(self.inner.as_ref())
     }
-    fn inner_mut(&mut self) -> Option<&mut dyn writer::Stackable<'a, Cookie>> {
+    fn inner_mut(&mut self) -> Option<&mut (dyn writer::Stackable<'a, Cookie> + Send + Sync)> {
         Some(self.inner.as_mut())
     }
     fn cookie_set(&mut self, cookie: Cookie) -> Cookie {
@@ -634,7 +634,7 @@ pub struct Signer<'a> {
     // take our inner reader.  If that happens, we only update the
     // digests.
     inner: Option<writer::BoxStack<'a, Cookie>>,
-    signers: