ipc: Make gnupg::KeyPair usable in async contexts.

  - See if we are executing under a tokio async runtime, and if so,
    start a new one on a different thread.

  - This works around a design problem with the
    openpgp::crypto::{Signer, Decryptor} traits that use sync functions,
    but our implementation of the traits is async.  We used to
    unconditionally start a tokio runtime and block to hide the async
    nature of the implementation, but that leads to panics if the
    current thread is already managed by a tokio runtime.  This is a
    really easy mistake to make, and is not detected by the type
    system.

4 files changed, 184 insertions, 10 deletions

diff --git a/Cargo.lock b/Cargo.lock
index 4ac9cbda..feae375b 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2723,6 +2723,7 @@ dependencies = [
  "buffered-reader",
  "capnp-rpc",
  "clap 3.2.19",
+ "crossbeam-utils",
  "ctor",
  "dirs",
  "fs2",
@@ -2950,7 +2951,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f31bf4e9fe5cd8cea8e0887e2e4eb1b4d736ff11b776c8537bf0912a4b381285"
 dependencies = [
  "digest 0.9.0",
- "generic-array 0.12.4",
+ "generic-array 0.14.5",
 ]
 
 [[package]]
diff --git a/ipc/Cargo.toml b/ipc/Cargo.toml
index dfe652ae..7ff590e1 100644
--- a/ipc/Cargo.toml
+++ b/ipc/Cargo.toml
@@ -26,6 +26,7 @@ sequoia-openpgp = { path = "../openpgp", version = "1.0.0", default-features = f
 anyhow = "1.0.18"
 buffered-reader = { path = "../buffered-reader", version = "1.0.0", default-features = false }
 capnp-rpc = "0.14"
+crossbeam-utils = "0.8"
 fs2 = "0.4.2"
 futures = "0.3.5"
 lalrpop-util = ">=0.17"
diff --git a/ipc/src/gnupg.rs b/ipc/src/gnupg.rs
index c6aa1468..0bec65ed 100644
--- a/ipc/src/gnupg.rs
+++ b/ipc/src/gnupg.rs
@@ -616,13 +616,38 @@ impl crypto::Signer for KeyPair {
                 | (DSA, PublicKey::DSA { .. })
                 | (EdDSA, PublicKey::EdDSA { .. })
                 | (ECDSA, PublicKey::ECDSA { .. }) => {
-                    let rt = tokio::runtime::Runtime::new()?;
+                    use tokio::runtime::{Handle, Runtime};
 
-                    rt.block_on(async move {
-                        let mut a = Agent::connect_to(&self.agent_socket).await?;
+                    // Connect to the agent and do the signing
+                    // operation.
+                    let do_it = async move {
+                        let mut a =
+                            Agent::connect_to(&self.agent_socket).await?;
                         let sig = a.sign(self, hash_algo, digest).await?;
                         Ok(sig)
-                    })
+                    };
+
+                    // See if the current thread is managed by a tokio
+                    // runtime.
+                    if Handle::try_current().is_err() {
+                        // Doesn't seem to be the case, so it is safe
+                        // to create a new runtime and block.
+                        let rt = Runtime::new()?;
+                        rt.block_on(do_it)
+                    } else {
+                        // It is!  We must not create a second runtime
+                        // on this thread, but instead we will
+                        // delegate this to a new thread.
+                        use crossbeam_utils::thread;
+
+                        thread::scope(|s| {
+                            s.spawn(move |_| {
+                                let rt = Runtime::new()?;
+                                rt.block_on(do_it)
+                            }).join()
+                        }).map_err(map_panic)?
+                            .map_err(map_panic)?
+                    }
                 },
 
             (pk_algo, _) => Err(openpgp::Error::InvalidOperation(format!(
@@ -647,13 +672,38 @@ impl crypto::Decryptor for KeyPair {
             (PublicKey::RSA { .. }, Ciphertext::RSA { .. })
                 | (PublicKey::ElGamal { .. }, Ciphertext::ElGamal { .. })
                 | (PublicKey::ECDH { .. }, Ciphertext::ECDH { .. }) => {
-                    let rt = tokio::runtime::Runtime::new()?;
+                    use tokio::runtime::{Handle, Runtime};
 
-                    rt.block_on(async move {
-                        let mut a = Agent::connect_to(&self.agent_socket).await?;
+                    // Connect to the agent and do the decryption
+                    // operation.
+                    let do_it = async move {
+                        let mut a =
+                            Agent::connect_to(&self.agent_socket).await?;
                         let sk = a.decrypt(self, ciphertext).await?;
                         Ok(sk)
-                    })
+                    };
+
+                    // See if the current thread is managed by a tokio
+                    // runtime.
+                    if Handle::try_current().is_err() {
+                        // Doesn't seem to be the case, so it is safe
+                        // to create a new runtime and block.
+                        let rt = Runtime::new()?;
+                        rt.block_on(do_it)
+                    } else {
+                        // It is!  We must not create a second runtime
+                        // on this thread, but instead we will
+                        // delegate this to a new thread.
+                        use crossbeam_utils::thread;
+
+                        thread::scope(|s| {
+                            s.spawn(move |_| {
+                                let rt = Runtime::new()?;
+                                rt.block_on(do_it)
+                            }).join()
+                        }).map_err(map_panic)?
+                            .map_err(map_panic)?
+                    }
                 },
 
             (public, ciphertext) =>
@@ -665,6 +715,14 @@ impl crypto::Decryptor for KeyPair {
     }
 }
 
+/// Maps a panic of a worker thread to an error.
+///
+/// Unfortunately, there is nothing useful to do with the error, but
+/// returning a generic error is better than panicking.
+fn map_panic(_: Box<dyn std::any::Any + std::marker::Send>) -> anyhow::Error
+{
+    anyhow::anyhow!("worker thread panicked")
+}
 
 #[derive(thiserror::Error, Debug)]
 /// Errors used in this module.
@@ -680,3 +738,91 @@ pub enum Error {
     ProtocolError(String),
 
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use openpgp::{
+        Cert,
+        crypto::{
+            hash::Digest,
+            mpi::Ciphertext,
+            Decryptor,
+            Signer,
+        },
+    };
+
+    /// Asserts that a <KeyPair as Signer> is usable from an async
+    /// context.
+    ///
+    /// Previously, the test died with
+    ///
+    ///     thread 'gnupg::tests::signer_in_async_context' panicked at
+    ///     'Cannot start a runtime from within a runtime. This
+    ///     happens because a function (like `block_on`) attempted to
+    ///     block the current thread while the thread is being used to
+    ///     drive asynchronous tasks.'
+    #[test]
+    fn signer_in_async_context() -> Result<()> {
+        async fn async_context() -> Result<()> {
+            let ctx = match Context::ephemeral() {
+                Ok(c) => c,
+                Err(e) => {
+                    eprintln!("Failed to create ephemeral context: {}", e);
+                    eprintln!("Most likely, GnuPG isn't installed.");
+                    eprintln!("Skipping test.");
+                    return Ok(());
+                },
+            };
+
+            let key = Cert::from_bytes(crate::tests::key("testy-new.pgp"))?
+                .primary_key().key().clone();
+            let mut pair = KeyPair::new(&ctx, &key)?;
+            let algo = HashAlgorithm::default();
+            let digest = algo.context()?.into_digest()?;
+            let _ = pair.sign(algo, &digest);
+            Ok(())
+        }
+
+        let rt = tokio::runtime::Runtime::new()?;
+        rt.block_on(async_context())
+    }
+
+    /// Asserts that a <KeyPair as Decryptor> is usable from an async
+    /// context.
+    ///
+    /// Previously, the test died with
+    ///
+    ///     thread 'gnupg::tests::decryptor_in_async_context' panicked
+    ///     at 'Cannot start a runtime from within a runtime. This
+    ///     happens because a function (like `block_on`) attempted to
+    ///     block the current thread while the thread is being used to
+    ///     drive asynchronous tasks.'
+    #[test]
+    fn decryptor_in_async_context() -> Result<()> {
+        async fn async_context() -> Result<()> {
+            let ctx = match Context::ephemeral() {
+                Ok(c) => c,
+                Err(e) => {
+                    eprintln!("Failed to create ephemeral context: {}", e);
+                    eprintln!("Most likely, GnuPG isn't installed.");
+                    eprintln!("Skipping test.");
+                    return Ok(());
+                },
+            };
+
+            let key = Cert::from_bytes(crate::tests::key("testy-new.pgp"))?
+                .keys().nth(1).unwrap().key().clone();
+            let mut pair = KeyPair::new(&ctx, &key)?;
+            let ciphertext = Ciphertext::ECDH {
+                e: vec![].into(),
+                key: vec![].into_boxed_slice(),
+            };
+            let _ = pair.decrypt(&ciphertext, None);
+            Ok(())
+        }
+
+        let rt = tokio::runtime::Runtime::new()?;
+        rt.block_on(async_context())
+    }
+}
diff --git a/ipc/tests/gpg-agent.rs b/ipc/tests/gpg-agent.rs
index a244ba44..331cc298 100644
--- a/ipc/tests/gpg-agent.rs
+++ b/ipc/tests/gpg-agent.rs
@@ -106,6 +106,18 @@ fn gpg_import(ctx: &Context, what: &[u8]) -> openpgp::Result<()> {
 }
 
 #[test]
+fn sync_sign() -> openpgp::Result<()> {
+    sign()
+}
+
+#[test]
+fn async_sign() -> openpgp::Result<()> {
+    let rt = tokio::runtime::Runtime::new()?;
+    rt.block_on(async {
+        sign()
+    })
+}
+
 fn sign() -> openpgp::Result<()> {
     use self::CipherSuite::*;
     use openpgp::policy::StandardPolicy as P;
@@ -215,7 +227,19 @@ fn sign() -> openpgp::Result<()> {
 }
 
 #[test]
-fn decrypt() -> openpgp::Result<()> {
+fn sync_decrypt() -> openpgp::Result<()> {
+    decrypt(true)
+}
+
+#[test]
+fn async_decrypt() -> openpgp::Result<()> {
+    let rt = tokio::runtime::Runtime::new()?;
+    rt.block_on(async {
+        decrypt(false)
+    })
+}
+
+fn decrypt(also_try_explicit_async: bool) -> openpgp::Result<()> {
     use self::CipherSuite::*;
     use openpgp::policy::StandardPolicy as P;
 
@@ -274,6 +298,7 @@ fn decrypt() -> openpgp::Result<()> {
             literal_writer.finalize().unwrap();
         }
 
+      if also_try_explicit_async {
         // First, test Agent::decrypt.  Using this function we can try
         // multiple decryption requests on the same connection.
         let rt = tokio::runtime::Runtime::new()?;
@@ -305,6 +330,7 @@ fn decrypt() -> openpgp::Result<()> {
 
         // Close connection.
         drop(agent);
+      }
 
         // Make a helper that that feeds the recipient's secret key to the
         // decryptor.