Merge branch 'vesoft-inc:master' into expression_depth_3

src/clients/meta/MetaClient.cpp

@@ -12,11 +12,13 @@
 #include <thrift/lib/cpp/util/EnumUtils.h>
 
 #include <boost/filesystem.hpp>
+#include <unordered_set>
 
 #include "clients/meta/FileBasedClusterIdMan.h"
 #include "clients/meta/stats/MetaClientStats.h"
 #include "common/base/Base.h"
 #include "common/base/MurmurHash2.h"
+#include "common/base/Status.h"
 #include "common/conf/Configuration.h"
 #include "common/http/HttpClient.h"
 #include "common/meta/NebulaSchemaProvider.h"
@@ -37,6 +39,27 @@ DEFINE_int32(meta_client_retry_interval_secs, 1, "meta client sleep interval bet
 DEFINE_int32(meta_client_timeout_ms, 60 * 1000, "meta client timeout");
 DEFINE_string(cluster_id_path, "cluster.id", "file path saved clusterId");
 DEFINE_int32(check_plan_killed_frequency, 8, "check plan killed every 1<<n times");
+DEFINE_uint32(failed_login_attempts,
+              0,
+              "how many consecutive incorrect passwords input to a SINGLE graph service node cause "
+              "the account to become locked.");
+DEFINE_uint32(
+    password_lock_time_in_secs,
+    0,
+    "how long in seconds to lock the account after too many consecutive login attempts provide an "
+    "incorrect password.");
+
+// Sanity-checking Flag Values
+static bool ValidateFailedLoginAttempts(const char* flagname, uint32_t value) {
+  if (value <= 32767)  // value is ok
+    return true;
+
+  FLOG_WARN("Invalid value for --%s: %d, the timeout should be an integer between 0 and 32767\n",
+            flagname,
+            (int)value);
+  return false;
+}
+DEFINE_validator(failed_login_attempts, &ValidateFailedLoginAttempts);
 
 namespace nebula {
 namespace meta {
@@ -152,6 +175,11 @@ bool MetaClient::loadUsersAndRoles() {
   }
   decltype(userRolesMap_) userRolesMap;
   decltype(userPasswordMap_) userPasswordMap;
+  decltype(userPasswordAttemptsRemain_) userPasswordAttemptsRemain;
+  decltype(userLoginLockTime_) userLoginLockTime;
+  // List of username
+  std::unordered_set<std::string> userNameList;
+
   for (auto& user : userRoleRet.value()) {
     auto rolesRet = getUserRoles(user.first).get();
     if (!rolesRet.ok()) {
@@ -160,11 +188,40 @@ bool MetaClient::loadUsersAndRoles() {
     }
     userRolesMap[user.first] = rolesRet.value();
     userPasswordMap[user.first] = user.second;
+    userPasswordAttemptsRemain[user.first] = FLAGS_failed_login_attempts;
+    userLoginLockTime[user.first] = 0;
+    userNameList.emplace(user.first);
   }
   {
     folly::RWSpinLock::WriteHolder holder(localCacheLock_);
     userRolesMap_ = std::move(userRolesMap);
     userPasswordMap_ = std::move(userPasswordMap);
+
+    // This method is called periodically by the heartbeat thread, but we don't want to reset the
+    // failed login attempts every time. Remove expired users from cache
+    for (auto& ele : userPasswordAttemptsRemain) {
+      if (userNameList.count(ele.first) == 0) {
+        userPasswordAttemptsRemain.erase(ele.first);
+      }
+    }
+    for (auto& ele : userLoginLockTime) {
+      if (userNameList.count(ele.first) == 0) {
+        userLoginLockTime.erase(ele.first);
+      }
+    }
+
+    // If the user is not in the map, insert value with the default value
+    for (const auto& user : userNameList) {
+      if (userPasswordAttemptsRemain.count(user) == 0) {
+        userPasswordAttemptsRemain[user] = FLAGS_failed_login_attempts;
+      }
+      if (userLoginLockTime.count(user) == 0) {
+        userLoginLockTime[user] = 0;
+      }
+    }
+
+    userPasswordAttemptsRemain_ = std::move(userPasswordAttemptsRemain);
+    userLoginLockTime_ = std::move(userLoginLockTime);
   }
   return true;
 }
@@ -694,6 +751,7 @@ void MetaClient::getResponse(Request req,
                   return;
                 } else {
                   LOG(ERROR) << "Send request to " << host << ", exceed retry limit";
+                  LOG(ERROR) << "RpcResponse exception: " << t.exception().what().c_str();
                   pro.setValue(
                       Status::Error("RPC failure in MetaClient: %s", t.exception().what().c_str()));
                 }
@@ -1369,6 +1427,8 @@ folly::Future<StatusOr<bool>> MetaClient::multiPut(
 
   cpp2::MultiPutReq req;
   std::vector<nebula::KeyValue> data;
+  data.reserve(pairs.size());
+
   for (auto& element : pairs) {
     data.emplace_back(std::move(element));
   }
@@ -2353,16 +2413,73 @@ std::vector<cpp2::RoleItem> MetaClient::getRolesByUserFromCache(const std::strin
   return iter->second;
 }
 
-bool MetaClient::authCheckFromCache(const std::string& account, const std::string& password) {
+Status MetaClient::authCheckFromCache(const std::string& account, const std::string& password) {
+  // Check meta service status
   if (!ready_) {
-    return false;
+    return Status::Error("Meta Service not ready");
   }
+
   const ThreadLocalInfo& threadLocalInfo = getThreadLocalInfo();
+  // Check user existence
   auto iter = threadLocalInfo.userPasswordMap_.find(account);
   if (iter == threadLocalInfo.userPasswordMap_.end()) {
-    return false;
+    return Status::Error("User not exist");
+  }
+
+  folly::RWSpinLock::WriteHolder holder(localCacheLock_);
+
+  auto& lockedSince = userLoginLockTime_[account];
+  auto& passwordAttemtRemain = userPasswordAttemptsRemain_[account];
+  LOG(INFO) << "Thread id: " << std::this_thread::get_id()
+            << " ,passwordAttemtRemain: " << passwordAttemtRemain;
+  // lockedSince is non-zero means the account has been locked
+  if (lockedSince != 0) {
+    auto remainingLockTime =
+        (lockedSince + FLAGS_password_lock_time_in_secs) - time::WallClock::fastNowInSec();
+    // If the account is still locked, there is no need to check the password
+    if (remainingLockTime > 0) {
+      return Status::Error(
+          "%d times consecutive incorrect passwords has been input, user name: %s has been "
+          "locked, try again in %ld seconds",
+          FLAGS_failed_login_attempts,
+          account.c_str(),
+          remainingLockTime);
+    }
+    // Clear lock state and reset attempts
+    lockedSince = 0;
+    passwordAttemtRemain = FLAGS_failed_login_attempts;
   }
-  return iter->second == password;
+
+  if (iter->second != password) {
+    // By default there is no limit of login attempts if any of these 2 flags is unset
+    if (FLAGS_failed_login_attempts == 0 || FLAGS_password_lock_time_in_secs == 0) {
+      return Status::Error("Invalid password");
+    }
+
+    // If the password is not correct and passwordAttemtRemain > 0,
+    // Allow another attemp
+    if (passwordAttemtRemain > 0) {
+      --passwordAttemtRemain;
+      if (passwordAttemtRemain == 0) {
+        // If the remaining attemps is 0, failed to authenticate
+        // Block user login
+        lockedSince = time::WallClock::fastNowInSec();
+        return Status::Error(
+            "%d times consecutive incorrect passwords has been input, user name: %s has been "
+            "locked, try again in %d seconds",
+            FLAGS_failed_login_attempts,
+            account.c_str(),
+            FLAGS_password_lock_time_in_secs);
+      }
+      LOG(ERROR) << "Invalid password, remaining attempts: " << passwordAttemtRemain;
+      return Status::Error("Invalid password, remaining attempts: %d", passwordAttemtRemain);
+    }
+  }
+
+  // Reset password attempts
+  passwordAttemtRemain = FLAGS_failed_login_attempts;
+  lockedSince = 0;
+  return Status::OK();
 }
 
 bool MetaClient::checkShadowAccountFromCache(const std::string& account) {
@@ -3532,6 +3649,21 @@ folly::Future<StatusOr<cpp2::ExecResp>> MetaClient::killQuery(
   return future;
 }
 
+folly::Future<StatusOr<int64_t>> MetaClient::getWorkerId(std::string ipAddr) {
+  cpp2::GetWorkerIdReq req;
+  req.host_ref() = std::move(ipAddr);
+
+  folly::Promise<StatusOr<int64_t>> promise;
+  auto future = promise.getFuture();
+  getResponse(
+      std::move(req),
+      [](auto client, auto request) { return client->future_getWorkerId(request); },
+      [](cpp2::GetWorkerIdResp&& resp) -> int64_t { return std::move(resp).get_workerid(); },
+      std::move(promise),
+      true);
+  return future;
+}
+
 folly::Future<StatusOr<bool>> MetaClient::download(const std::string& hdfsHost,
                                                    int32_t hdfsPort,
                                                    const std::string& hdfsPath,

src/clients/meta/MetaClient.h

@@ -14,6 +14,7 @@
 #include <gtest/gtest_prod.h>
 
 #include <atomic>
+#include <cstdint>
 
 #include "common/base/Base.h"
 #include "common/base/ObjectPool.h"
@@ -143,6 +144,10 @@ using IndexStatus = std::tuple<std::string, std::string, std::string>;
 using UserRolesMap = std::unordered_map<std::string, std::vector<cpp2::RoleItem>>;
 // get user password by account
 using UserPasswordMap = std::unordered_map<std::string, std::string>;
+// Mapping of user name and remaining wrong password attempts
+using UserPasswordAttemptsRemain = std::unordered_map<std::string, uint32>;
+// Mapping of user name and the timestamp when the account is locked
+using UserLoginLockTime = std::unordered_map<std::string, uint32>;
 
 // config cache, get config via module and name
 using MetaConfigMap =
@@ -199,13 +204,13 @@ struct MetaClientOptions {
   std::string serviceName_ = "";
   // Whether to skip the config manager
   bool skipConfig_ = false;
-  // host role(graph/meta/storage) using this client
+  // Host role(graph/meta/storage) using this client
   cpp2::HostRole role_ = cpp2::HostRole::UNKNOWN;
   // gitInfoSHA of Host using this client
   std::string gitInfoSHA_{""};
-  // data path list, used in storaged
+  // Data path list, used in storaged
   std::vector<std::string> dataPaths_;
-  // install path, used in metad/graphd/storaged
+  // Install path, used in metad/graphd/storaged
   std::string rootPath_;
 };
 
@@ -593,7 +598,7 @@ class MetaClient {
 
   std::vector<cpp2::RoleItem> getRolesByUserFromCache(const std::string& user);
 
-  bool authCheckFromCache(const std::string& account, const std::string& password);
+  Status authCheckFromCache(const std::string& account, const std::string& password);
 
   StatusOr<TermID> getTermFromCache(GraphSpaceID spaceId, PartitionID);
 
@@ -651,6 +656,8 @@ class MetaClient {
 
   folly::Future<StatusOr<bool>> ingest(GraphSpaceID spaceId);
 
+  folly::Future<StatusOr<int64_t>> getWorkerId(std::string ipAddr);
+
   HostAddr getMetaLeader() {
     return leader_;
   }
@@ -659,6 +666,10 @@ class MetaClient {
     return heartbeatTime_;
   }
 
+  std::string getLocalIp() {
+    return options_.localHost_.toString();
+  }
+
  protected:
   // Return true if load succeeded.
   bool loadData();
@@ -815,6 +826,8 @@ class MetaClient {
 
   UserRolesMap userRolesMap_;
   UserPasswordMap userPasswordMap_;
+  UserPasswordAttemptsRemain userPasswordAttemptsRemain_;
+  UserLoginLockTime userLoginLockTime_;
 
   NameIndexMap tagNameIndexMap_;
   NameIndexMap edgeNameIndexMap_;

src/clients/storage/StorageClient.cpp

@@ -485,6 +485,7 @@ StorageRpcRespFuture<cpp2::LookupIndexResp> StorageClient::lookupIndex(
     bool isEdge,
     int32_t tagOrEdge,
     const std::vector<std::string>& returnCols,
+    std::vector<storage::cpp2::OrderBy> orderBy,
     int64_t limit) {
   // TODO(sky) : instead of isEdge and tagOrEdge to nebula::cpp2::SchemaID for graph layer.
   auto space = param.space;
@@ -516,6 +517,7 @@ StorageRpcRespFuture<cpp2::LookupIndexResp> StorageClient::lookupIndex(
     req.indices_ref() = spec;
     req.common_ref() = common;
     req.limit_ref() = limit;
+    req.order_by_ref() = orderBy;
   }
 
   return collectResponse(param.evb,

src/clients/storage/StorageClient.h

@@ -141,6 +141,7 @@ class StorageClient
       bool isEdge,
       int32_t tagOrEdge,
       const std::vector<std::string>& returnCols,
+      std::vector<storage::cpp2::OrderBy> orderBy,
       int64_t limit);
 
   StorageRpcRespFuture<cpp2::GetNeighborsResponse> lookupAndTraverse(

src/common/CMakeLists.txt

@@ -27,3 +27,4 @@ nebula_add_subdirectory(utils)
 nebula_add_subdirectory(ssl)
 nebula_add_subdirectory(geo)
 nebula_add_subdirectory(memory)
+nebula_add_subdirectory(id)

src/common/expression/UUIDExpression.cpp

@@ -10,36 +10,26 @@
 namespace nebula {
 
 bool UUIDExpression::operator==(const Expression& rhs) const {
-  if (kind_ != rhs.kind()) {
-    return false;
-  }
-
-  const auto& r = static_cast<const UUIDExpression&>(rhs);
-  return field_ == r.field_;
+  return kind_ == rhs.kind();
 }
 
 void UUIDExpression::writeTo(Encoder& encoder) const {
   // kind_
+  UNUSED(encoder);
   encoder << kind_;
-
-  // field_
-  CHECK(!field_.empty());
-  encoder << field_;
 }
 
 void UUIDExpression::resetFrom(Decoder& decoder) {
-  // Read field_
-  field_ = decoder.readStr();
+  UNUSED(decoder);
 }
 
 const Value& UUIDExpression::eval(ExpressionContext& ctx) {
-  // TODO
   UNUSED(ctx);
   return result_;
 }
 
 std::string UUIDExpression::toString() const {
-  return folly::stringPrintf("uuid(\"%s\")", field_.c_str());
+  return folly::stringPrintf("uuid()");
 }
 
 void UUIDExpression::accept(ExprVisitor* visitor) {

src/common/expression/UUIDExpression.h

@@ -14,8 +14,8 @@ class UUIDExpression final : public Expression {
   friend class Expression;
 
  public:
-  static UUIDExpression* make(ObjectPool* pool, const std::string& field = "") {
-    return pool->add(new UUIDExpression(pool, field));
+  static UUIDExpression* make(ObjectPool* pool) {
+    return pool->add(new UUIDExpression(pool));
   }
 
   bool operator==(const Expression& rhs) const override;
@@ -27,18 +27,16 @@ class UUIDExpression final : public Expression {
   void accept(ExprVisitor* visitor) override;
 
   Expression* clone() const override {
-    return UUIDExpression::make(pool_, field_);
+    return UUIDExpression::make(pool_);
   }
 
  private:
-  explicit UUIDExpression(ObjectPool* pool, const std::string& field = "")
-      : Expression(pool, Kind::kUUID), field_(field) {}
+  explicit UUIDExpression(ObjectPool* pool) : Expression(pool, Kind::kUUID) {}
 
   void writeTo(Encoder& encoder) const override;
   void resetFrom(Decoder& decoder) override;
 
  private:
-  std::string field_;
   Value result_;
 };
 

src/common/expression/test/EncodeDecodeTest.cpp

@@ -205,7 +205,7 @@ TEST(ExpressionEncodeDecode, TypeCastingExpression) {
 }
 
 TEST(ExpressionEncodeDecode, UUIDExpression) {
-  const auto& uuidEx = *UUIDExpression::make(&pool, "field");
+  const auto& uuidEx = *UUIDExpression::make(&pool);
   std::string encoded = Expression::encode(uuidEx);
   auto decoded = Expression::decode(&pool, folly::StringPiece(encoded.data(), encoded.size()));
   EXPECT_EQ(uuidEx, *decoded);