Reject unbalanced and overly deep type names

clickhouse/types/type_parser.cpp

@@ -209,16 +209,19 @@ bool TypeParser::Parse(TypeAst* type) {
                 type_->value_string = std::string(token.value);
                 break;
             case Token::LPar:
+                if (open_elements_.size() > kMaxNestingDepth) return false;
                 type_->elements.emplace_back(TypeAst());
                 open_elements_.push(type_);
                 type_ = &type_->elements.back();
                 break;
             case Token::RPar:
+                if (open_elements_.size() < 2) return false;
                 type_ = open_elements_.top();
                 open_elements_.pop();
                 break;
             case Token::Assign:
             case Token::Comma:
+                if (open_elements_.size() < 2) return false;
                 type_ = open_elements_.top();
                 open_elements_.pop();
                 type_->elements.emplace_back(TypeAst());

clickhouse/types/type_parser.h

@@ -71,6 +71,11 @@ class TypeParser {
     };
 
 public:
+    // Current servers reject column types nested deeper than 499 levels. The
+    // column factory and TypeAst destruction recurse once per level, so the
+    // limit also bounds their stack use.
+    static constexpr size_t kMaxNestingDepth = 512;
+
     explicit TypeParser(const StringView& name);
     ~TypeParser();
 

ut/type_parser_ut.cpp

@@ -1,3 +1,4 @@
+#include <clickhouse/columns/factory.h>
 #include <clickhouse/types/type_parser.h>
 #include <gtest/gtest.h>
 
@@ -552,3 +553,75 @@ TEST(TypeParser, AggregateFunction) {
         EXPECT_FALSE(TypeParser("AggregateFunction(argMax, LowCardinality(Nullable(FixedString(4))), DateTime(3, 'UTC'))").Parse(&ast));
     }
 }
+
+TEST(TypeParser, UnbalancedClosingParenthesis) {
+    for (const std::string type_name : {")", "),", ")))", "UInt8)", "UInt8))", "UInt8),", "UInt8)(", "Array(UInt8))", "Array(UInt8)),",
+                                        "Tuple(UInt8, String))", "UInt8,", "UInt8, String", ","}) {
+        TypeAst ast;
+        EXPECT_FALSE(TypeParser(type_name).Parse(&ast)) << type_name;
+        EXPECT_EQ(nullptr, ParseTypeName(type_name)) << type_name;
+    }
+}
+
+namespace {
+std::string NestedArrayType(size_t depth, bool closed = true) {
+    std::string result;
+    for (size_t i = 0; i < depth; ++i) {
+        result += "Array(";
+    }
+    result += "UInt8";
+    if (closed) {
+        result.append(depth, ')');
+    }
+    return result;
+}
+}  // namespace
+
+TEST(TypeParser, NestingDepthLimit) {
+    {
+        TypeAst ast;
+        ASSERT_TRUE(TypeParser(NestedArrayType(TypeParser::kMaxNestingDepth)).Parse(&ast));
+
+        size_t depth        = 0;
+        const TypeAst* node = &ast;
+        while (node->meta == TypeAst::Array) {
+            ASSERT_EQ(1u, node->elements.size());
+            node = &node->elements.front();
+            ++depth;
+        }
+        EXPECT_EQ(TypeParser::kMaxNestingDepth, depth);
+        EXPECT_EQ(Type::UInt8, node->code);
+    }
+
+    for (size_t depth : {TypeParser::kMaxNestingDepth + 1, size_t{100000}}) {
+        TypeAst closed_ast;
+        EXPECT_FALSE(TypeParser(NestedArrayType(depth)).Parse(&closed_ast)) << depth;
+        EXPECT_EQ(nullptr, ParseTypeName(NestedArrayType(depth))) << depth;
+
+        TypeAst open_ast;
+        EXPECT_FALSE(TypeParser(NestedArrayType(depth, false)).Parse(&open_ast)) << depth;
+        EXPECT_EQ(nullptr, ParseTypeName(NestedArrayType(depth, false))) << depth;
+    }
+}
+
+TEST(TypeParser, NestingDepthLimitBoundsColumnFactory) {
+    auto col = CreateColumnByType(NestedArrayType(32));
+    ASSERT_NE(nullptr, col);
+    EXPECT_EQ(NestedArrayType(32), col->Type()->GetName());
+
+    EXPECT_NE(nullptr, CreateColumnByType(NestedArrayType(499)));
+    EXPECT_NE(nullptr, CreateColumnByType(NestedArrayType(TypeParser::kMaxNestingDepth)));
+    EXPECT_EQ(nullptr, CreateColumnByType(NestedArrayType(TypeParser::kMaxNestingDepth + 1)));
+    EXPECT_EQ(nullptr, CreateColumnByType(NestedArrayType(100000)));
+    EXPECT_EQ(nullptr, CreateColumnByType(NestedArrayType(100000, false)));
+}
+
+TEST(TypeParser, BalancedNestedControls) {
+    for (const std::string type_name : {"Map(String, Array(Nullable(UInt8)))", "Tuple(a UInt8, b String)",
+                                        "Array(Array(Array(Array(Array(Array(Array(Array(UInt8))))))))"}) {
+        TypeAst ast;
+        EXPECT_TRUE(TypeParser(type_name).Parse(&ast)) << type_name;
+        EXPECT_NE(nullptr, ParseTypeName(type_name)) << type_name;
+        EXPECT_NE(nullptr, CreateColumnByType(type_name)) << type_name;
+    }
+}