http: refactor headersTimeout and requestTimeout logic

PR-URL: https://github.com/nodejs/node/pull/41263
Fixes: https://github.com/nodejs/node/issues/33440
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Robert Nagy <ronagy@icloud.com>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Darshan Sen <raisinten@gmail.com>

src/node_http_parser.cc

@@ -60,6 +60,7 @@ using v8::FunctionTemplate;
 using v8::HandleScope;
 using v8::Int32;
 using v8::Integer;
+using v8::Isolate;
 using v8::Local;
 using v8::MaybeLocal;
 using v8::Number;
@@ -187,7 +188,58 @@ struct StringPtr {
   size_t size_;
 };
 
+class Parser;
+
+struct ParserComparator {
+  bool operator()(const Parser* lhs, const Parser* rhs) const;
+};
+
+class ConnectionsList : public BaseObject {
+ public:
+    static void New(const FunctionCallbackInfo<Value>& args);
+
+    static void All(const FunctionCallbackInfo<Value>& args);
+
+    static void Idle(const FunctionCallbackInfo<Value>& args);
+
+    static void Active(const FunctionCallbackInfo<Value>& args);
+
+    static void Expired(const FunctionCallbackInfo<Value>& args);
+
+    void Push(Parser* parser) {
+      all_connections_.insert(parser);
+    }
+
+    void Pop(Parser* parser) {
+      all_connections_.erase(parser);
+    }
+
+    void PushActive(Parser* parser) {
+      active_connections_.insert(parser);
+    }
+
+    void PopActive(Parser* parser) {
+      active_connections_.erase(parser);
+    }
+
+    SET_NO_MEMORY_INFO()
+    SET_MEMORY_INFO_NAME(ConnectionsList)
+    SET_SELF_SIZE(ConnectionsList)
+
+ private:
+    ConnectionsList(Environment* env, Local<Object> object)
+      : BaseObject(env, object) {
+        MakeWeak();
+      }
+
+    std::set<Parser*, ParserComparator> all_connections_;
+    std::set<Parser*, ParserComparator> active_connections_;
+};
+
 class Parser : public AsyncWrap, public StreamListener {
+  friend class ConnectionsList;
+  friend struct ParserComparator;
+
  public:
   Parser(BindingData* binding_data, Local<Object> wrap)
       : AsyncWrap(binding_data->env(), wrap),
@@ -205,10 +257,21 @@ class Parser : public AsyncWrap, public StreamListener {
   SET_SELF_SIZE(Parser)
 
   int on_message_begin() {
+    // Important: Pop from the list BEFORE resetting the last_message_start_
+    // otherwise std::set.erase will fail.
+    if (connectionsList_ != nullptr) {
+      connectionsList_->PopActive(this);
+    }
+
     num_fields_ = num_values_ = 0;
+    headers_completed_ = false;
+    last_message_start_ = uv_hrtime();
     url_.Reset();
     status_message_.Reset();
-    header_parsing_start_time_ = uv_hrtime();
+
+    if (connectionsList_ != nullptr) {
+      connectionsList_->PushActive(this);
+    }
 
     Local<Value> cb = object()->Get(env()->context(), kOnMessageBegin)
                               .ToLocalChecked();
@@ -297,8 +360,8 @@ class Parser : public AsyncWrap, public StreamListener {
 
 
   int on_headers_complete() {
+    headers_completed_ = true;
     header_nread_ = 0;
-    header_parsing_start_time_ = 0;
 
     // Arguments for the on-headers-complete javascript callback. This
     // list needs to be kept in sync with the actual argument list for
@@ -429,6 +492,14 @@ class Parser : public AsyncWrap, public StreamListener {
   int on_message_complete() {
     HandleScope scope(env()->isolate());
 
+    // Important: Pop from the list BEFORE resetting the last_message_start_
+    // otherwise std::set.erase will fail.
+    if (connectionsList_ != nullptr) {
+      connectionsList_->PopActive(this);
+    }
+
+    last_message_start_ = 0;
+
     if (num_fields_)
       Flush();  // Flush trailing HTTP headers.
 
@@ -486,6 +557,11 @@ class Parser : public AsyncWrap, public StreamListener {
     Parser* parser;
     ASSIGN_OR_RETURN_UNWRAP(&parser, args.Holder());
 
+    if (parser->connectionsList_ != nullptr) {
+      parser->connectionsList_->Pop(parser);
+      parser->connectionsList_->PopActive(parser);
+    }
+
     // Since the Parser destructor isn't going to run the destroy() callbacks
     // it needs to be triggered manually.
     parser->EmitTraceEventDestroy();
@@ -506,7 +582,6 @@ class Parser : public AsyncWrap, public StreamListener {
     }
   }
 
-
   // var bytesParsed = parser->execute(buffer);
   static void Execute(const FunctionCallbackInfo<Value>& args) {
     Parser* parser;
@@ -545,8 +620,8 @@ class Parser : public AsyncWrap, public StreamListener {
     Environment* env = Environment::GetCurrent(args);
 
     uint64_t max_http_header_size = 0;
-    uint64_t headers_timeout = 0;
     uint32_t lenient_flags = kLenientNone;
+    ConnectionsList* connectionsList = nullptr;
 
     CHECK(args[0]->IsInt32());
     CHECK(args[1]->IsObject());
@@ -565,9 +640,9 @@ class Parser : public AsyncWrap, public StreamListener {
       lenient_flags = args[3].As<Int32>()->Value();
     }
 
-    if (args.Length() > 4) {
-      CHECK(args[4]->IsInt32());
-      headers_timeout = args[4].As<Int32>()->Value();
+    if (args.Length() > 4 && !args[4]->IsNullOrUndefined()) {
+      CHECK(args[4]->IsObject());
+      ASSIGN_OR_RETURN_UNWRAP(&connectionsList, args[4]);
     }
 
     llhttp_type_t type =
@@ -586,7 +661,21 @@ class Parser : public AsyncWrap, public StreamListener {
 
     parser->set_provider_type(provider);
     parser->AsyncReset(args[1].As<Object>());
-    parser->Init(type, max_http_header_size, lenient_flags, headers_timeout);
+    parser->Init(type, max_http_header_size, lenient_flags);
+
+    if (connectionsList != nullptr) {
+      parser->connectionsList_ = connectionsList;
+
+      parser->connectionsList_->Push(parser);
+
+      // This protects from a DoS attack where an attacker establishes
+      // the connection without sending any data on applications where
+      // server.timeout is left to the default value of zero.
+      parser->last_message_start_ = uv_hrtime();
+      parser->connectionsList_->PushActive(parser);
+    } else {
+      parser->connectionsList_ = nullptr;
+    }
   }
 
   template <bool should_pause>
@@ -644,6 +733,26 @@ class Parser : public AsyncWrap, public StreamListener {
     args.GetReturnValue().Set(ret);
   }
 
+  static void Duration(const FunctionCallbackInfo<Value>& args) {
+    Parser* parser;
+    ASSIGN_OR_RETURN_UNWRAP(&parser, args.Holder());
+
+    if (parser->last_message_start_ == 0) {
+      args.GetReturnValue().Set(0);
+      return;
+    }
+
+    double duration = (uv_hrtime() - parser->last_message_start_) / 1e6;
+    args.GetReturnValue().Set(duration);
+  }
+
+  static void HeadersCompleted(const FunctionCallbackInfo<Value>& args) {
+    Parser* parser;
+    ASSIGN_OR_RETURN_UNWRAP(&parser, args.Holder());
+
+    args.GetReturnValue().Set(parser->headers_completed_);
+  }
+
  protected:
   static const size_t kAllocBufferSize = 64 * 1024;
 
@@ -690,24 +799,6 @@ class Parser : public AsyncWrap, public StreamListener {
     if (ret.IsEmpty())
       return;
 
-    // check header parsing time
-    if (header_parsing_start_time_ != 0 && headers_timeout_ != 0) {
-      uint64_t now = uv_hrtime();
-      uint64_t parsing_time = (now - header_parsing_start_time_) / 1000000;
-
-      if (parsing_time > headers_timeout_) {
-        Local<Value> cb =
-            object()->Get(env()->context(), kOnTimeout).ToLocalChecked();
-
-        if (!cb->IsFunction())
-          return;
-
-        MakeCallback(cb.As<Function>(), 0, nullptr);
-
-        return;
-      }
-    }
-
     Local<Value> cb =
         object()->Get(env()->context(), kOnExecute).ToLocalChecked();
 
@@ -853,7 +944,7 @@ class Parser : public AsyncWrap, public StreamListener {
 
 
   void Init(llhttp_type_t type, uint64_t max_http_header_size,
-            uint32_t lenient_flags, uint64_t headers_timeout) {
+            uint32_t lenient_flags) {
     llhttp_init(&parser_, type, &settings);
 
     if (lenient_flags & kLenientHeaders) {
@@ -873,9 +964,8 @@ class Parser : public AsyncWrap, public StreamListener {
     num_values_ = 0;
     have_flushed_ = false;
     got_exception_ = false;
+    headers_completed_ = false;
     max_http_header_size_ = max_http_header_size;
-    header_parsing_start_time_ = 0;
-    headers_timeout_ = headers_timeout;
   }
 
 
@@ -923,11 +1013,12 @@ class Parser : public AsyncWrap, public StreamListener {
   size_t current_buffer_len_;
   const char* current_buffer_data_;
   unsigned int execute_depth_ = 0;
+  bool headers_completed_ = false;
   bool pending_pause_ = false;
   uint64_t header_nread_ = 0;
   uint64_t max_http_header_size_;
-  uint64_t headers_timeout_;
-  uint64_t header_parsing_start_time_ = 0;
+  uint64_t last_message_start_;
+  ConnectionsList* connectionsList_;
 
   BaseObjectPtr<BindingData> binding_data_;
 
@@ -952,6 +1043,135 @@ class Parser : public AsyncWrap, public StreamListener {
   static const llhttp_settings_t settings;
 };
 
+bool ParserComparator::operator()(const Parser* lhs, const Parser* rhs) const {
+  if (lhs->last_message_start_ == 0) {
+    return false;
+  } else if (rhs->last_message_start_ == 0) {
+    return true;
+  }
+
+  return lhs->last_message_start_ < rhs->last_message_start_;
+}
+
+void ConnectionsList::New(const FunctionCallbackInfo<Value>& args) {
+  Local<Context> context = args.GetIsolate()->GetCurrentContext();
+  Environment* env = Environment::GetCurrent(context);
+
+  new ConnectionsList(env, args.This());
+}
+
+void ConnectionsList::All(const FunctionCallbackInfo<Value>& args) {
+  Isolate* isolate = args.GetIsolate();
+  Local<Context> context = isolate->GetCurrentContext();
+
+  Local<Array> all = Array::New(isolate);
+  ConnectionsList* list;
+
+  ASSIGN_OR_RETURN_UNWRAP(&list, args.Holder());
+
+  uint32_t i = 0;
+  for (auto parser : list->all_connections_) {
+    if (all->Set(context, i++, parser->object()).IsNothing()) {
+      return;
+    }
+  }
+
+  return args.GetReturnValue().Set(all);
+}
+
+void ConnectionsList::Idle(const FunctionCallbackInfo<Value>& args) {
+  Isolate* isolate = args.GetIsolate();
+  Local<Context> context = isolate->GetCurrentContext();
+
+  Local<Array> idle = Array::New(isolate);
+  ConnectionsList* list;
+
+  ASSIGN_OR_RETURN_UNWRAP(&list, args.Holder());
+
+  uint32_t i = 0;
+  for (auto parser : list->all_connections_) {
+    if (parser->last_message_start_ == 0) {
+      if (idle->Set(context, i++, parser->object()).IsNothing()) {
+        return;
+      }
+    }
+  }
+
+  return args.GetReturnValue().Set(idle);
+}
+
+void ConnectionsList::Active(const FunctionCallbackInfo<Value>& args) {
+  Isolate* isolate = args.GetIsolate();
+  Local<Context> context = isolate->GetCurrentContext();
+
+  Local<Array> active = Array::New(isolate);
+  ConnectionsList* list;
+
+  ASSIGN_OR_RETURN_UNWRAP(&list, args.Holder());
+
+  uint32_t i = 0;
+  for (auto parser : list->active_connections_) {
+    if (active->Set(context, i++, parser->object()).IsNothing()) {
+      return;
+    }
+  }
+
+  return args.GetReturnValue().Set(active);
+}
+
+void ConnectionsList::Expired(const FunctionCallbackInfo<Value>& args) {
+  Isolate* isolate = args.GetIsolate();
+  Local<Context> context = isolate->GetCurrentContext();
+
+  Local<Array> expired = Array::New(isolate);
+  ConnectionsList* list;
+
+  ASSIGN_OR_RETURN_UNWRAP(&list, args.Holder());
+  CHECK(args[0]->IsNumber());
+  CHECK(args[1]->IsNumber());
+  uint64_t headers_timeout =
+    static_cast<uint64_t>(args[0].As<Uint32>()->Value()) * 1000000;
+  uint64_t request_timeout =
+    static_cast<uint64_t>(args[1].As<Uint32>()->Value()) * 1000000;
+
+  if (headers_timeout == 0 && request_timeout == 0) {
+    return args.GetReturnValue().Set(expired);
+  } else if (request_timeout > 0 && headers_timeout > request_timeout) {
+    std::swap(headers_timeout, request_timeout);
+  }
+
+  const uint64_t now = uv_hrtime();
+  const uint64_t headers_deadline =
+    headers_timeout > 0 ? now - headers_timeout : 0;
+  const uint64_t request_deadline =
+    request_timeout > 0 ? now - request_timeout : 0;
+
+  uint32_t i = 0;
+  auto iter = list->active_connections_.begin();
+  auto end = list->active_connections_.end();
+  while (iter != end) {
+    Parser* parser = *iter;
+    iter++;
+
+    // Check for expiration.
+    if (
+      (!parser->headers_completed_ && headers_deadline > 0 &&
+        parser->last_message_start_ < headers_deadline) ||
+      (
+        request_deadline > 0 &&
+        parser->last_message_start_ < request_deadline)
+    ) {
+      if (expired->Set(context, i++, parser->object()).IsNothing()) {
+        return;
+      }
+
+      list->active_connections_.erase(parser);
+    }
+  }
+
+  return args.GetReturnValue().Set(expired);
+}
+
 const llhttp_settings_t Parser::settings = {
   Proxy<Call, &Parser::on_message_begin>::Raw,
   Proxy<DataCall, &Parser::on_url>::Raw,
@@ -1038,8 +1258,19 @@ void InitializeHttpParser(Local<Object> target,
   env->SetProtoMethod(t, "consume", Parser::Consume);
   env->SetProtoMethod(t, "unconsume", Parser::Unconsume);
   env->SetProtoMethod(t, "getCurrentBuffer", Parser::GetCurrentBuffer);
+  env->SetProtoMethod(t, "duration", Parser::Duration);
+  env->SetProtoMethod(t, "headersCompleted", Parser::HeadersCompleted);
 
   env->SetConstructorFunction(target, "HTTPParser", t);
+
+  Local<FunctionTemplate> c = env->NewFunctionTemplate(ConnectionsList::New);
+  c->InstanceTemplate()
+    ->SetInternalFieldCount(ConnectionsList::kInternalFieldCount);
+  env->SetProtoMethod(c, "all", ConnectionsList::All);
+  env->SetProtoMethod(c, "idle", ConnectionsList::Idle);
+  env->SetProtoMethod(c, "active", ConnectionsList::Active);
+  env->SetProtoMethod(c, "expired", ConnectionsList::Expired);
+  env->SetConstructorFunction(target, "ConnectionsList", c);
 }
 
 }  // anonymous namespace