[#2583] Added connection filter callback

src/lib/tcp/tcp_connection.*
    TcpConnection added connection_filter callback invoked in acceptorCallback()
    prior to initiating handshake.

src/lib/tcp/tcp_listener.*
    Added connection filter callback to ctor, which is passed into createConnection()
    and used duing connection acceptance to potentially reject connections

src/lib/tcp/tests/tcp_listener_unittests.cc
    Added filter test

src/lib/tcp/tests/tcp_messages.*
    Added new log message

9 files changed, 146 insertions, 30 deletions

diff --git a/src/lib/tcp/tcp_connection.cc b/src/lib/tcp/tcp_connection.cc
index 8c23280665..aa90f9f86a 100644
--- a/src/lib/tcp/tcp_connection.cc
+++ b/src/lib/tcp/tcp_connection.cc
@@ -52,7 +52,8 @@ TcpConnection::TcpConnection(asiolink::IOService& io_service,
                              const TcpConnectionAcceptorPtr& acceptor,
                              const TlsContextPtr& tls_context,
                              TcpConnectionPool& connection_pool,
-                             const TcpConnectionAcceptorCallback& callback,
+                             const TcpConnectionAcceptorCallback& acceptor_callback,
+                             const TcpConnectionFilterCallback& connection_filter,
                              const long idle_timeout,
                              const size_t read_max /* = 32768 */)
     : tls_context_(tls_context),
@@ -62,7 +63,8 @@ TcpConnection::TcpConnection(asiolink::IOService& io_service,
       tls_socket_(),
       acceptor_(acceptor),
       connection_pool_(connection_pool),
-      acceptor_callback_(callback),
+      acceptor_callback_(acceptor_callback),
+      connection_filter_(connection_filter),
       input_buf_(read_max) {
     if (!tls_context) {
         tcp_socket_.reset(new asiolink::TCPSocket<SocketCallback>(io_service));
@@ -282,9 +284,18 @@ TcpConnection::acceptorCallback(const boost::system::error_code& ec) {
         stopThisConnection();
     }
 
+    // Stage a new connection to listen for next client.
     acceptor_callback_(ec);
 
     if (!ec) {
+        if (!(connection_filter_(getRemoteEndpointAddressAsText()))) {
+            LOG_DEBUG(tcp_logger, isc::log::DBGLVL_TRACE_DETAIL,
+                      TCP_CONNECTION_REJECTED_BY_FILTER)
+                      .arg(getRemoteEndpointAddressAsText());
+            stopThisConnection();
+            return;
+        }
+
         if (!tls_context_) {
             LOG_DEBUG(tcp_logger, isc::log::DBGLVL_TRACE_DETAIL,
                       TCP_REQUEST_RECEIVE_START)
diff --git a/src/lib/tcp/tcp_connection.h b/src/lib/tcp/tcp_connection.h
index 811e1d0503..b21483a545 100644
--- a/src/lib/tcp/tcp_connection.h
+++ b/src/lib/tcp/tcp_connection.h
@@ -158,6 +158,9 @@ public:
 /// declaring @ref TcpConnectionPool to avoid circular inclusion.
 class TcpConnectionPool;
 
+/// @brief Type of the callback for filtering new connections by ip address.
+typedef std::function<bool(const std::string& remote_endpoint_address)> TcpConnectionFilterCallback;
+
 /// @brief Accepts and handles a single TCP connection.
 class TcpConnection : public boost::enable_shared_from_this<TcpConnection> {
 private:
@@ -207,7 +210,9 @@ public:
     /// @param tls_context TLS context.
     /// @param connection_pool Connection pool in which this connection is
     /// stored.
-    /// @param callback Callback invoked when new connection is accepted.
+    /// @param acceptor_callback Callback invoked when new connection is accepted.
+    /// @param connection_filter Callback invoked prior to handshake which can be
+    /// used to qualify and reject connections
     /// @param idle_timeout Timeout after which a TCP connection is
     /// closed by the server.
     /// @param read_max maximum size of a single socket read.  Defaults to 32K.
@@ -215,7 +220,8 @@ public:
                   const TcpConnectionAcceptorPtr& acceptor,
                   const asiolink::TlsContextPtr& tls_context,
                   TcpConnectionPool& connection_pool,
-                  const TcpConnectionAcceptorCallback& callback,
+                  const TcpConnectionAcceptorCallback& acceptor_callback,
+                  const TcpConnectionFilterCallback& connection_filter,
                   const long idle_timeout,
                   const size_t read_max = 32768);
 
@@ -427,6 +433,9 @@ protected:
     /// @brief External TCP acceptor callback.
     TcpConnectionAcceptorCallback acceptor_callback_;
 
+    /// @brief External callback for filtering connections by IP address.
+    TcpConnectionFilterCallback connection_filter_;
+
     /// @brief Maximum bytes to read in a single socket read.
     size_t read_max_;
 
diff --git a/src/lib/tcp/tcp_listener.cc b/src/lib/tcp/tcp_listener.cc
index 4a6543dd27..31a6f8b2a5 100644
--- a/src/lib/tcp/tcp_listener.cc
+++ b/src/lib/tcp/tcp_listener.cc
@@ -18,10 +18,11 @@ TcpListener::TcpListener(IOService& io_service,
                          const IOAddress& server_address,
                          const unsigned short server_port,
                          const TlsContextPtr& tls_context,
-                         const IdleTimeout& idle_timeout)
+                         const IdleTimeout& idle_timeout,
+                         const TcpConnectionFilterCallback& connection_filter)
     : io_service_(io_service), tls_context_(tls_context), acceptor_(),
-      endpoint_(), connections_(), idle_timeout_(idle_timeout.value_) {
-
+      endpoint_(), connections_(), idle_timeout_(idle_timeout.value_),
+      connection_filter_(connection_filter) {
     // Create the TCP or TLS acceptor.
     if (!tls_context) {
         acceptor_.reset(new TcpConnectionAcceptor(io_service));
@@ -77,7 +78,7 @@ TcpListener::accept() {
     TcpConnectionAcceptorCallback acceptor_callback =
         std::bind(&TcpListener::acceptHandler, this, ph::_1);
 
-    TcpConnectionPtr conn = createConnection(acceptor_callback);
+    TcpConnectionPtr conn = createConnection(acceptor_callback, connection_filter_);
 
     // Add this new connection to the pool.
     connections_.start(conn);
@@ -91,7 +92,8 @@ TcpListener::acceptHandler(const boost::system::error_code&) {
 }
 
 TcpConnectionPtr
-TcpListener::createConnection(const TcpConnectionAcceptorCallback& /* callback */) {
+TcpListener::createConnection(const TcpConnectionAcceptorCallback&,
+                              const TcpConnectionFilterCallback&) {
     isc_throw(NotImplemented, "TcpListener::createConnection:");
 }
 
diff --git a/src/lib/tcp/tcp_listener.h b/src/lib/tcp/tcp_listener.h
index b3e6621a20..8086399e9b 100644
--- a/src/lib/tcp/tcp_listener.h
+++ b/src/lib/tcp/tcp_listener.h
@@ -51,7 +51,8 @@ public:
     /// @param server_port Port number on which the TCP service should run.
     /// @param tls_context TLS context.
     /// @param idle_timeout Timeout after which an idle TCP connection is
-    /// closed by the server.
+    /// @param connection_filter Callback invoked during connection acceptance
+    /// that can allow or deny connections based on the remote endpoint.
     ///
     /// @throw TcpListenerError when any of the specified parameters is
     /// invalid.
@@ -59,7 +60,8 @@ public:
                 const asiolink::IOAddress& server_address,
                 const unsigned short server_port,
                 const asiolink::TlsContextPtr& tls_context,
-                const IdleTimeout& idle_timeout);
+                const IdleTimeout& idle_timeout,
+                const TcpConnectionFilterCallback& connection_filter);
 
     /// @brief Virtual destructor.
     virtual ~TcpListener() {
@@ -109,8 +111,13 @@ protected:
     /// This method is virtual so as it can be overridden when customized
     /// connections are to be used, e.g. in case of unit testing.
     ///
+    /// @param acceptor_callback Callback invoked when new connection is accepted.
+    /// @param connection_filter Callback invoked during acceptance which may
+    /// allow of deny connections based on their remote address.
     /// @return Pointer to the created connection.
-    virtual TcpConnectionPtr createConnection(const TcpConnectionAcceptorCallback& callback);
+    virtual TcpConnectionPtr createConnection(
+        const TcpConnectionAcceptorCallback& acceptor_callback,
+        const TcpConnectionFilterCallback& connection_filter);
 
     /// @brief Reference to the IO service.
     asiolink::IOService& io_service_;
@@ -131,6 +138,10 @@ protected:
     /// @brief Timeout after which idle connection is closed by
     /// the server.
     long idle_timeout_;
+
+    /// @brief Callback invoked during acceptance which may
+    /// reject connections.
+    TcpConnectionFilterCallback connection_filter_;
 };
 
 } // end of namespace isc::asiolink
diff --git a/src/lib/tcp/tcp_messages.cc b/src/lib/tcp/tcp_messages.cc
index 34ad8560d5..85f4a210b7 100644
--- a/src/lib/tcp/tcp_messages.cc
+++ b/src/lib/tcp/tcp_messages.cc
@@ -11,6 +11,7 @@ extern const isc::log::MessageID TCP_CLIENT_REQUEST_RECEIVED = "TCP_CLIENT_REQUE
 extern const isc::log::MessageID TCP_CONNECTION_CLOSE_CALLBACK_FAILED = "TCP_CONNECTION_CLOSE_CALLBACK_FAILED";
 extern const isc::log::MessageID TCP_CONNECTION_HANDSHAKE_FAILED = "TCP_CONNECTION_HANDSHAKE_FAILED";
 extern const isc::log::MessageID TCP_CONNECTION_HANDSHAKE_START = "TCP_CONNECTION_HANDSHAKE_START";
+extern const isc::log::MessageID TCP_CONNECTION_REJECTED_BY_FILTER = "TCP_CONNECTION_REJECTED_BY_FILTER";
 extern const isc::log::MessageID TCP_CONNECTION_SHUTDOWN = "TCP_CONNECTION_SHUTDOWN";
 extern const isc::log::MessageID TCP_CONNECTION_SHUTDOWN_FAILED = "TCP_CONNECTION_SHUTDOWN_FAILED";
 extern const isc::log::MessageID TCP_CONNECTION_STOP = "TCP_CONNECTION_STOP";
@@ -33,6 +34,7 @@ const char* values[] = {
     "TCP_CONNECTION_CLOSE_CALLBACK_FAILED", "Connection close callback threw an exception",
     "TCP_CONNECTION_HANDSHAKE_FAILED", "TLS handshake with %1 failed with %2",
     "TCP_CONNECTION_HANDSHAKE_START", "start TLS handshake with %1 with timeout %2",
+    "TCP_CONNECTION_REJECTED_BY_FILTER", "connection from %1 has been denied by the connection filter.",
     "TCP_CONNECTION_SHUTDOWN", "shutting down TCP connection from %1",
     "TCP_CONNECTION_SHUTDOWN_FAILED", "shutting down TCP connection failed",
     "TCP_CONNECTION_STOP", "stopping TCP connection from %1",
diff --git a/src/lib/tcp/tcp_messages.h b/src/lib/tcp/tcp_messages.h
index 91e7aa3fae..bde8b12ac3 100644
--- a/src/lib/tcp/tcp_messages.h
+++ b/src/lib/tcp/tcp_messages.h
@@ -12,6 +12,7 @@ extern const isc::log::MessageID TCP_CLIENT_REQUEST_RECEIVED;
 extern const isc::log::MessageID TCP_CONNECTION_CLOSE_CALLBACK_FAILED;
 extern const isc::log::MessageID TCP_CONNECTION_HANDSHAKE_FAILED;
 extern const isc::log::MessageID TCP_CONNECTION_HANDSHAKE_START;
+extern const isc::log::MessageID TCP_CONNECTION_REJECTED_BY_FILTER;
 extern const isc::log::MessageID TCP_CONNECTION_SHUTDOWN;
 extern const isc::log::MessageID TCP_CONNECTION_SHUTDOWN_FAILED;
 extern const isc::log::MessageID TCP_CONNECTION_STOP;
diff --git a/src/lib/tcp/tcp_messages.mes b/src/lib/tcp/tcp_messages.mes
index 756536ea9b..523930ca42 100644
--- a/src/lib/tcp/tcp_messages.mes
+++ b/src/lib/tcp/tcp_messages.mes
@@ -61,6 +61,10 @@ transaction. This is proven to occur when the system clock is moved manually
 or as a result of synchronization with a time server. Any ongoing transactions
 will be interrupted. New transactions should be conducted normally.
 
+% TCP_CONNECTION_REJECTED_BY_FILTER connection from %1 has been denied by the connection filter.
+This debug message is issued when the server's connection filter rejects
+a new connection based on the client's ip address.
+
 % TCP_REQUEST_RECEIVE_START start receiving request from %1 with timeout %2
 This debug message is issued when the server starts receiving new request
 over the established connection. The first argument specifies the address
@@ -73,7 +77,6 @@ server attempted to process a received request. The first argument specifies
 the address of the remote endpoint. The second argument describes the nature
 error.
 
-
 % TCP_SERVER_RESPONSE_SEND sending TCP response %1 to %2
 This debug message is issued when the server is starting to send a TCP
 response to a remote endpoint. The first argument holds basic information
diff --git a/src/lib/tcp/tests/tcp_listener_unittests.cc b/src/lib/tcp/tests/tcp_listener_unittests.cc
index c69f051d9e..e472bc06b6 100644
--- a/src/lib/tcp/tests/tcp_listener_unittests.cc
+++ b/src/lib/tcp/tests/tcp_listener_unittests.cc
@@ -69,10 +69,11 @@ public:
                       const TcpConnectionAcceptorPtr& acceptor,
                       const TlsContextPtr& tls_context,
                       TcpConnectionPool& connection_pool,
-                      const TcpConnectionAcceptorCallback& callback,
+                      const TcpConnectionAcceptorCallback& acceptor_callback,
+                      const TcpConnectionFilterCallback& filter_callback,
                       const long idle_timeout)
-     : TcpConnection(io_service, acceptor, tls_context, connection_pool, callback,
-                     idle_timeout) {
+     : TcpConnection(io_service, acceptor, tls_context, connection_pool,
+                     acceptor_callback, filter_callback, idle_timeout) {
     }
 
     virtual TcpRequestPtr createRequest() {
@@ -112,9 +113,10 @@ public:
                     const unsigned short server_port,
                     const TlsContextPtr& tls_context,
                     const IdleTimeout& idle_timeout,
+                    const TcpConnectionFilterCallback& filter_callback,
                     const size_t read_max = 32 * 1024)
         : TcpListener(io_service, server_address, server_port,
-                      tls_context, idle_timeout),
+                      tls_context, idle_timeout, filter_callback),
                       read_max_(read_max) {
     }
 
@@ -127,18 +129,19 @@ protected:
     ///
     /// @param callback Callback invoked when new connection is accepted.
     /// @return Pointer to the created connection.
-    virtual TcpConnectionPtr createConnection(const TcpConnectionAcceptorCallback& callback) {
-        TcpConnectionPtr
-            conn(new TcpTestConnection(io_service_, acceptor_,
-                                       tls_context_, connections_,
-                                       callback, idle_timeout_));
-            conn->setReadMax(read_max_);
+    virtual TcpConnectionPtr createConnection(
+            const TcpConnectionAcceptorCallback& acceptor_callback,
+            const TcpConnectionFilterCallback& connection_filter) {
+        TcpConnectionPtr conn(new TcpTestConnection(io_service_, acceptor_,
+                                                    tls_context_, connections_,
+                                                    acceptor_callback, connection_filter,
+                                                    idle_timeout_));
+        conn->setReadMax(read_max_);
         return (conn);
     }
 
     /// @brief Maximum size of a single socket read
     size_t read_max_;
-
 };
 
 /// @brief Test fixture class for @ref TcpListener.
@@ -240,6 +243,28 @@ public:
         io_service_.poll();
     }
 
+    /// @brief Pass through filter that allows all connections.
+    bool noFilter(const std::string& /* remote_endpoint_address */) {
+        return(true);
+    }
+
+    /// @brief Filter that denies every other connection.
+    ///
+    /// @param remote_endpoint_address ip address of the remote end of
+    /// a connection.
+    bool connectionFilter(const std::string& remote_endpoint_address) {
+        static size_t count = 0;
+        // If the address doesn't match, something hinky is going on, so
+        // we'll reject them all.  If it does match, then cool, it works
+        // as expected.
+        if ((count++ % 2) || (remote_endpoint_address != SERVER_ADDRESS)) {
+            // Reject every other connection;
+            return (false);
+        }
+
+        return (true);
+    }
+
     /// @brief IO service used in the tests.
     IOService io_service_;
 
@@ -263,7 +288,8 @@ TEST_F(TcpListenerTest, listen) {
     const std::string request = "I am done";
 
     TcpTestListener listener(io_service_, IOAddress(SERVER_ADDRESS), SERVER_PORT,
-                             TlsContextPtr(), TcpListener::IdleTimeout(IDLE_TIMEOUT));
+                             TlsContextPtr(), TcpListener::IdleTimeout(IDLE_TIMEOUT),
+                             std::bind(&TcpListenerTest::noFilter, this, ph::_1));
 
     ASSERT_NO_THROW(listener.start());
     ASSERT_EQ(SERVER_ADDRESS, listener.getLocalAddress().toText());
@@ -288,7 +314,9 @@ TEST_F(TcpListenerTest, splitReads) {
     // Read at most one byte at a time.
     size_t read_max = 1;
     TcpTestListener listener(io_service_, IOAddress(SERVER_ADDRESS), SERVER_PORT,
-                             TlsContextPtr(), TcpListener::IdleTimeout(IDLE_TIMEOUT), read_max);
+                             TlsContextPtr(), TcpListener::IdleTimeout(IDLE_TIMEOUT),
+                             std::bind(&TcpListenerTest::noFilter, this, ph::_1),
+                             read_max);
 
     ASSERT_NO_THROW(listener.start());
     ASSERT_EQ(SERVER_ADDRESS, listener.getLocalAddress().toText());
@@ -311,7 +339,8 @@ TEST_F(TcpListenerTest, splitReads) {
 // transmit a streamed request and receive a streamed response.
 TEST_F(TcpListenerTest, idleTimeoutTest) {
     TcpTestListener listener(io_service_, IOAddress(SERVER_ADDRESS), SERVER_PORT,
-                             TlsContextPtr(), TcpListener::IdleTimeout(SHORT_IDLE_TIMEOUT));
+                             TlsContextPtr(), TcpListener::IdleTimeout(SHORT_IDLE_TIMEOUT),
+                             std::bind(&TcpListenerTest::noFilter, this, ph::_1));
 
     ASSERT_NO_THROW(listener.start());
     ASSERT_EQ(SERVER_ADDRESS, listener.getLocalAddress().toText());
@@ -337,7 +366,8 @@ TEST_F(TcpListenerTest, multipleClientsListen) {
     const std::string request = "I am done";
 
     TcpTestListener listener(io_service_, IOAddress(SERVER_ADDRESS), SERVER_PORT,
-                             TlsContextPtr(), TcpListener::IdleTimeout(IDLE_TIMEOUT));
+                             TlsContextPtr(), TcpListener::IdleTimeout(IDLE_TIMEOUT),
+                             std::bind(&TcpListenerTest::noFilter, this, ph::_1));
 
     ASSERT_NO_THROW(listener.start());
     ASSERT_EQ(SERVER_ADDRESS, listener.getLocalAddress().toText());
@@ -359,11 +389,14 @@ TEST_F(TcpListenerTest, multipleClientsListen) {
     io_service_.poll();
 }
 
+// Verify that the listener handles multiple requests for multiple
+// clients.
 TEST_F(TcpListenerTest, multipleRequetsPerClients) {
     std::list<std::string>requests{ "one", "two", "three", "I am done"};
 
     TcpTestListener listener(io_service_, IOAddress(SERVER_ADDRESS), SERVER_PORT,
-                             TlsContextPtr(), TcpListener::IdleTimeout(IDLE_TIMEOUT));
+                             TlsContextPtr(), TcpListener::IdleTimeout(IDLE_TIMEOUT),
+                             std::bind(&TcpListenerTest::noFilter, this, ph::_1));
 
     ASSERT_NO_THROW(listener.start());
     ASSERT_EQ(SERVER_ADDRESS, listener.getLocalAddress().toText());
@@ -387,4 +420,48 @@ TEST_F(TcpListenerTest, multipleRequetsPerClients) {
     io_service_.poll();
 }
 
+// Verify that connection filtering can eliminate specific connections.
+TEST_F(TcpListenerTest, filterClientsTest) {
+    const std::string request = "I am done";
+
+    TcpTestListener listener(io_service_, IOAddress(SERVER_ADDRESS), SERVER_PORT,
+                             TlsContextPtr(), TcpListener::IdleTimeout(IDLE_TIMEOUT),
+                             std::bind(&TcpListenerTest::connectionFilter, this, ph::_1));
+
+    ASSERT_NO_THROW(listener.start());
+    ASSERT_EQ(SERVER_ADDRESS, listener.getLocalAddress().toText());
+    ASSERT_EQ(SERVER_PORT, listener.getLocalPort());
+    size_t num_clients = 5;
+    for ( auto i = 0; i < num_clients; ++i ) {
+        // Every other client sends nothing (i.e. waits for EOF) as
+        // we expect the filter to reject them.
+        if (i % 2 == 0) {
+            ASSERT_NO_THROW(startRequest("I am done"));
+        } else {
+            ASSERT_NO_THROW(startRequest(""));
+        }
+    }
+
+    ASSERT_NO_THROW(runIOService());
+    ASSERT_EQ(num_clients, clients_.size());
+
+    size_t i = 0;
+    for (auto client = clients_.begin(); client != clients_.end(); ++client) {
+        if (i % 2 == 0) {
+            // These clients should have been accepted and received responses.
+            EXPECT_TRUE((*client)->receiveDone());
+            EXPECT_FALSE((*client)->expectedEof());
+        } else {
+            // These clients should have been rejected and gotten EOF'd.
+            EXPECT_FALSE((*client)->receiveDone());
+            EXPECT_TRUE((*client)->expectedEof());
+        }
+
+        ++i;
+    }
+
+    listener.stop();
+    io_service_.poll();
+}
+
 }
diff --git a/src/lib/tcp/tests/tcp_test_client.h b/src/lib/tcp/tests/tcp_test_client.h
index e3ea57c946..b02962eac2 100644
--- a/src/lib/tcp/tests/tcp_test_client.h
+++ b/src/lib/tcp/tests/tcp_test_client.h
@@ -245,7 +245,7 @@ public:
                     return;
                 } else {
                     // Error occurred, bail...
-                    ADD_FAILURE() << "error occurred while receiving TCP"
+                    ADD_FAILURE() << "client: " << this << " error occurred while receiving TCP"
                         " response from the server: " << ec.message();
                     done_callback_();
                     return;