Issue 29, check for bad client version

src/daemon/absorber.cc

@@ -13,6 +13,17 @@ using namespace std::placeholders;
 namespace dist_clang {
 namespace daemon {
 
+namespace {
+
+bool ValidateClientVersion(ui32 client_version) {
+  return client_version >= Absorber::kMinGoodClientVersion;
+}
+
+}  // namespace
+
+// static
+const ui32 Absorber::kMinGoodClientVersion = 100;
+
 Absorber::Absorber(const proto::Configuration& configuration)
     : CompilationDaemon(configuration) {
   using Worker = base::WorkerPool::SimpleWorker;
@@ -67,6 +78,14 @@ bool Absorber::HandleNewMessage(net::ConnectionPtr connection,
 
   if (message->HasExtension(proto::Remote::extension)) {
     Message execute(message->ReleaseExtension(proto::Remote::extension));
+    if (!ValidateClientVersion(execute->client_version())) {
+      LOG(WARNING) << "Client sent a bad version: "
+                   << execute->client_version();
+      net::proto::Status bad_version_status;
+      bad_version_status.set_code(net::proto::Status::BAD_CLIENT_VERSION);
+      return connection->ReportStatus(bad_version_status);
+    }
+
     DCHECK(!execute->flags().compiler().has_path());
     if (execute->has_source()) {
       return tasks_->Push(Task{connection, std::move(execute)});

src/daemon/absorber.h

@@ -9,6 +9,8 @@ namespace daemon {
 
 class Absorber : public CompilationDaemon {
  public:
+  static const ui32 kMinGoodClientVersion;
+
   explicit Absorber(const proto::Configuration& configuration);
   virtual ~Absorber();
 

src/daemon/absorber_test.cc

@@ -10,13 +10,16 @@ namespace {
 
 // S1..4 types can be one of the following: |String|, |Immutable|, |Literal|
 template <typename S1, typename S2, typename S3, typename S4 = String>
-net::Connection::ScopedMessage CreateMessage(const S1& source,
-                                             const S2& action,
-                                             const S3& compiler_version,
-                                             const S4& language = S4()) {
+net::Connection::ScopedMessage CreateMessage(
+    const S1& source,
+    const S2& action,
+    const S3& compiler_version,
+    const S4& language = S4(),
+    ui32 client_version = Absorber::kMinGoodClientVersion) {
   net::Connection::ScopedMessage message(new net::Connection::Message);
   auto* extension = message->MutableExtension(proto::Remote::extension);
   extension->set_source(source);
+  extension->set_client_version(client_version);
   auto* compiler = extension->mutable_flags()->mutable_compiler();
   compiler->set_version(compiler_version);
   extension->mutable_flags()->set_action(action);
@@ -374,5 +377,88 @@ TEST_F(AbsorberTest, BadMessageStatus) {
       << "Daemon must not store references to the connection";
 }
 
+TEST_F(AbsorberTest, BadClientVersion) {
+  const String expected_host = "fake_host";
+  const ui16 expected_port = 12345;
+  const String compiler_version("compiler_version");
+  const String compiler_path("compiler_path");
+
+  conf.mutable_absorber()->mutable_local()->set_host(expected_host);
+  conf.mutable_absorber()->mutable_local()->set_port(expected_port);
+  auto* version = conf.add_versions();
+  version->set_version(compiler_version);
+  version->set_path(compiler_path);
+
+  listen_callback =
+      [&expected_host, expected_port](const String& host, ui16 port, String*) {
+    EXPECT_EQ(expected_host, host);
+    EXPECT_EQ(expected_port, port);
+    return true;
+  };
+  connect_callback = [](net::TestConnection* connection) {
+    connection->CallOnSend([](const net::Connection::Message& message) {
+      EXPECT_TRUE(message.HasExtension(net::proto::Status::extension));
+      const auto& status = message.GetExtension(net::proto::Status::extension);
+      EXPECT_EQ(net::proto::Status::BAD_CLIENT_VERSION, status.code());
+
+      EXPECT_FALSE(message.HasExtension(proto::Result::extension));
+    });
+  };
+
+  absorber.reset(new Absorber(conf));
+  ASSERT_TRUE(absorber->Initialize());
+
+  auto connection1 = test_service->TriggerListen(expected_host, expected_port);
+  {
+    SharedPtr<net::TestConnection> test_connection =
+        std::static_pointer_cast<net::TestConnection>(connection1);
+
+    auto message(CreateMessage(""_l, ""_l, ""_l, ""_l, 1));
+    EXPECT_TRUE(
+        test_connection->TriggerReadAsync(std::move(message), StatusOK()));
+
+    UniqueLock lock(send_mutex);
+    ASSERT_TRUE(send_condition.wait_for(lock, std::chrono::seconds(1),
+                                        [this] { return send_count == 1; }));
+  }
+
+  auto connection2 = test_service->TriggerListen(expected_host, expected_port);
+  {
+    SharedPtr<net::TestConnection> test_connection =
+        std::static_pointer_cast<net::TestConnection>(connection2);
+
+    test_connection->CallOnSend([](const net::Connection::Message& message) {
+      EXPECT_TRUE(message.HasExtension(net::proto::Status::extension));
+      const auto& status = message.GetExtension(net::proto::Status::extension);
+      EXPECT_EQ(net::proto::Status::OK, status.code());
+
+      EXPECT_TRUE(message.HasExtension(proto::Result::extension));
+      EXPECT_TRUE(message.GetExtension(proto::Result::extension).has_obj());
+    });
+    auto message(CreateMessage("source"_l, "action"_l, compiler_version));
+    auto* extension = message->MutableExtension(proto::Remote::extension);
+    extension->clear_client_version();
+    EXPECT_TRUE(
+        test_connection->TriggerReadAsync(std::move(message), StatusOK()));
+
+    UniqueLock lock(send_mutex);
+    ASSERT_TRUE(send_condition.wait_for(lock, std::chrono::seconds(1),
+                                        [this] { return send_count == 2; }));
+  }
+
+  absorber.reset();
+
+  EXPECT_EQ(1u, run_count);
+  EXPECT_EQ(1u, listen_count);
+  EXPECT_EQ(2u, connect_count);
+  EXPECT_EQ(2u, connections_created);
+  EXPECT_EQ(2u, read_count);
+  EXPECT_EQ(2u, send_count);
+  EXPECT_EQ(1, connection1.use_count())
+      << "Daemon must not store references to the connection";
+  EXPECT_EQ(1, connection2.use_count())
+      << "Daemon must not store references to the connection";
+}
+
 }  // namespace daemon
 }  // namespace dist_clang

src/daemon/remote.proto

@@ -7,6 +7,7 @@ package dist_clang.daemon.proto;
 message Remote {
   optional base.proto.Flags flags = 1;
   optional string source          = 2;
+  optional uint32 client_version  = 3 [ default = 100 ];
 
   extend net.proto.Universal {
     optional Remote extension = 6;

src/net/universal.proto

@@ -17,6 +17,7 @@ message Status {
     EXECUTION     = 5;
     OVERLOAD      = 6;
     NO_VERSION    = 7;
+    BAD_CLIENT_VERSION = 8;
   }
 
   required Code code           = 1 [ default = OK ];