Add more tests for Pipe/Filter

In this round of write some tests and find a bug, Threaded_Fork
seems to be completely broken. I don't think the semaphore approach
it uses really works (consistently) because a single worker thread
can acquire the semaphore more than once. This can be seen in the
(disabled) test of Threaded_Fork.

Not sure what to do about Threaded_Fork - it has been broken since
introduction and nobody has mentioned any problems so likely nobody has
ever used it. May actually be better to remove it entirely rather than
to fix it.

2 files changed, 172 insertions, 16 deletions

diff --git a/src/lib/filters/threaded_fork.cpp b/src/lib/filters/threaded_fork.cpp
index f558ac9c2..ff54bcbc6 100644
--- a/src/lib/filters/threaded_fork.cpp
+++ b/src/lib/filters/threaded_fork.cpp
@@ -136,6 +136,11 @@ void Threaded_Fork::thread_entry(Filter* filter)
    {
    while(true)
       {
+      /*
+      * This is plain wrong: a single thread can get the semaphore
+      * more than one time, meaning it will process the input twice
+      * and some other thread/filter will not see this input.
+      */
       m_thread_data->m_input_ready_semaphore.acquire();
 
       if(!m_thread_data->m_input)
diff --git a/src/tests/test_filters.cpp b/src/tests/test_filters.cpp
index b6bbf05c3..392fd98ef 100644
--- a/src/tests/test_filters.cpp
+++ b/src/tests/test_filters.cpp
@@ -1,5 +1,6 @@
 /*
 * (C) 2016 Daniel Neus
+*     2016 Jack Lloyd
 *
 * Botan is released under the Simplified BSD License (see license.txt)
 */
@@ -7,7 +8,9 @@
 #include "tests.h"
 
 #if defined(BOTAN_HAS_FILTERS)
-    #include <botan/secqueue.h>
+  #include <botan/secqueue.h>
+  #include <botan/pipe.h>
+  #include <botan/filters.h>
 #endif
 
 namespace Botan_Tests {
@@ -20,43 +23,191 @@ class Filter_Tests : public Test
       std::vector<Test::Result> run() override
          {
          std::vector<Test::Result> results;
-         Test::Result secqueue_result("SecureQueue");
+
+         results.push_back(test_secqueue());
+         results.push_back(test_pipe_hash());
+         results.push_back(test_pipe_stream());
+         results.push_back(test_pipe_codec());
+         results.push_back(test_fork());
+
+#if defined(BOTAN_TARGET_OS_HAS_THREADS) && 0
+         // Threaded_Fork is broken
+         results.push_back(test_threaded_fork());
+#endif
+
+         return results;
+         }
+
+   private:
+      Test::Result test_secqueue()
+         {
+         Test::Result result("SecureQueue");
 
          try
             {
-            using Botan::SecureQueue;
-            SecureQueue queue_a;
+            Botan::SecureQueue queue_a;
             std::vector<uint8_t> test_data = {0x24, 0xB2, 0xBF, 0xC2, 0xE6, 0xD4, 0x7E, 0x04, 0x67, 0xB3};
             queue_a.write(test_data.data(), test_data.size());
 
-            secqueue_result.test_eq("size of SecureQueue is correct", queue_a.size(), test_data.size());
-            secqueue_result.test_eq("0 bytes read so far from SecureQueue", queue_a.get_bytes_read(), 0);
+            result.test_eq("size of SecureQueue is correct", queue_a.size(), test_data.size());
+            result.test_eq("0 bytes read so far from SecureQueue", queue_a.get_bytes_read(), 0);
 
             uint8_t b;
             size_t bytes_read = queue_a.read_byte(b);
-            secqueue_result.test_eq("1 byte read", bytes_read, 1);
+            result.test_eq("1 byte read", bytes_read, 1);
 
             Botan::secure_vector<uint8_t> produced(b);
             Botan::secure_vector<uint8_t> expected(test_data.at(0));
-            secqueue_result.test_eq("byte read is correct", produced, expected);
+            result.test_eq("byte read is correct", produced, expected);
 
-            secqueue_result.test_eq("1 bytes read so far from SecureQueue", queue_a.get_bytes_read(), 1);
+            result.test_eq("1 bytes read so far from SecureQueue", queue_a.get_bytes_read(), 1);
 
-            SecureQueue queue_b;
+            Botan::SecureQueue queue_b;
             queue_a = queue_b;
-            secqueue_result.test_eq("bytes_read is set correctly", queue_a.get_bytes_read(), 0);
+            result.test_eq("bytes_read is set correctly", queue_a.get_bytes_read(), 0);
             }
          catch (std::exception& e)
             {
-            secqueue_result.test_failure("SecureQueue", e.what());
+            result.test_failure("SecureQueue", e.what());
             }
 
-         results.push_back(secqueue_result);
-         return results;
-         }	
+         return result;
+         }
+
+      Test::Result test_pipe_hash()
+         {
+         Test::Result result("Pipe");
+         Botan::Pipe pipe(new Botan::Hash_Filter("SHA-224"));
+         pipe.pop();
+         pipe.append(new Botan::Hash_Filter("SHA-256"));
+
+         result.test_eq("Message count", pipe.message_count(), 0);
+
+         pipe.start_msg();
+         uint8_t inb = 0x41;
+         pipe.write(&inb, 1);
+         pipe.write(std::vector<uint8_t>(6, 0x41));
+         pipe.write(inb);
+         pipe.end_msg();
+
+         result.test_eq("Message count", pipe.message_count(), 1);
+         result.test_eq("Message size", pipe.remaining(), 32);
+
+         std::vector<uint8_t> out(32);
+         result.test_eq("Expected read count", pipe.read(&out[0], 5), 5);
+         result.test_eq("Expected read count", pipe.read(&out[5], 17), 17);
+         result.test_eq("Expected read count", pipe.read(&out[22], 12), 10);
+         result.test_eq("Expected read count", pipe.read(&out[0], 1), 0); // no more output
+
+         result.test_eq("Expected output", out, "C34AB6ABB7B2BB595BC25C3B388C872FD1D575819A8F55CC689510285E212385");
+
+         return result;
+         }
+
+      Test::Result test_pipe_codec()
+         {
+         Test::Result result("Pipe");
+
+         Botan::Pipe pipe(new Botan::Base64_Encoder);
+
+         result.test_eq("Message count", pipe.message_count(), 0);
+
+         pipe.process_msg("ABCDX");
+
+         result.test_eq("Message count", pipe.message_count(), 1);
+         result.test_eq("Message size", pipe.remaining(), 8);
+
+         std::string output = pipe.read_all_as_string(0);
+         result.test_eq("Message size", pipe.remaining(0), 0);
+         result.test_eq("Output round tripped", output, "QUJDRFg=");
+
+         pipe.append(new Botan::Base64_Decoder);
+         pipe.process_msg("FOOBAZ");
+
+         result.test_eq("base64 roundtrip", pipe.read_all_as_string(1), "FOOBAZ");
+
+         pipe.pop();
+         pipe.pop();
+
+         // Pipe is empty of filters, should still pass through
+         pipe.process_msg("surprise plaintext");
+
+         pipe.set_default_msg(2);
+         result.test_eq("Message 2", pipe.read_all_as_string(), "surprise plaintext");
+
+         pipe.append(new Botan::Hex_Decoder);
+
+         pipe.process_msg("F331F00D");
+         Botan::secure_vector<uint8_t> bin = pipe.read_all(3);
+         result.test_eq("hex decoded", bin, "F331F00D");
+
+         pipe.append(new Botan::Hex_Encoder);
+         pipe.process_msg("F331F00D");
+         result.test_eq("hex roundtrip", pipe.read_all_as_string(4), "F331F00D");
+
+         return result;
+         }
+
+      Test::Result test_pipe_stream()
+         {
+         Test::Result result("Pipe");
+
+         Botan::Keyed_Filter* aes = nullptr;
+         const Botan::SymmetricKey key("FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF");
+         const Botan::InitializationVector iv("FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF");
+         Botan::Pipe pipe(aes = new Botan::StreamCipher_Filter("CTR-BE(AES-128)", key));
+
+         aes->set_iv(iv);
+
+         pipe.process_msg("ABCDEF");
+
+         result.test_eq("Message count", pipe.message_count(), 1);
+         result.test_eq("Ciphertext", pipe.read_all(), "FDFD6238F7C6");
+         return result;
+         }
+
+      Test::Result test_fork()
+         {
+         Test::Result result("Fork");
+
+         Botan::Pipe pipe(new Botan::Fork(new Botan::Hash_Filter("SHA-256"),
+                                          new Botan::Hash_Filter("SHA-512-256")));
+
+         result.test_eq("Message count", pipe.message_count(), 0);
+         pipe.process_msg("OMG");
+         result.test_eq("Message count", pipe.message_count(), 2);
+
+         // Test reading out of order
+         result.test_eq("Hash 2", pipe.read_all(1), "610480FFA82F24F6926544B976FE387878E3D973C03DFD591C2E9896EFB903E0");
+         result.test_eq("Hash 1", pipe.read_all(0), "C00862D1C6C1CF7C1B49388306E7B3C1BB79D8D6EC978B41035B556DBB3797DF");
+
+         return result;
+
+         }
+
+#if defined(BOTAN_TARGET_OS_HAS_THREADS)
+      Test::Result test_threaded_fork()
+         {
+         Test::Result result("Threaded_Fork");
+
+         Botan::Pipe pipe(new Botan::Threaded_Fork(new Botan::Hex_Encoder,
+                                                   new Botan::Base64_Encoder));
+
+         result.test_eq("Message count", pipe.message_count(), 0);
+         pipe.process_msg("woo");
+         result.test_eq("Message count", pipe.message_count(), 2);
+
+         // Test reading out of order
+         result.test_eq("Hash 2", pipe.read_all_as_string(1), "d29v");
+         result.test_eq("Hash 1", pipe.read_all_as_string(0), "776F6F");
+
+         return result;
+         }
+#endif
+
    };
 
-   BOTAN_REGISTER_TEST("filter", Filter_Tests);
+BOTAN_REGISTER_TEST("filter", Filter_Tests);
 
 #endif