fix(dynamodb): reject redundant condition-expression parentheses

moto/dynamodb/comparisons.py

@@ -4,7 +4,10 @@
 from decimal import Decimal
 from typing import Any, Optional, Union
 
-from moto.dynamodb.exceptions import ConditionAttributeIsReservedKeyword
+from moto.dynamodb.exceptions import (
+    ConditionAttributeIsReservedKeyword,
+    InvalidConditionExpression,
+)
 from moto.dynamodb.models.dynamo_type import Item
 from moto.dynamodb.parsing.reserved_keywords import ReservedKeywords
 
@@ -215,6 +218,7 @@ def parse(self) -> Union[Op, "Func"]:
         nodes = self._apply_between(nodes)
         nodes = self._apply_parens_and_booleans(nodes)
         node = nodes[0]
+        self._assert_no_redundant_parentheses(node)
 
         self.expr_attr_names_found.extend(self._find_literals(node))
         self.expr_attr_values_found.extend(self._find_expr_attr_values(node))
@@ -891,6 +895,30 @@ def _make_op_condition(self, node: Node) -> Union["Func", Op]:
         else:  # pragma: no cover
             raise ValueError(f"Unknown expression node kind {node.kind}")
 
+    def _assert_no_redundant_parentheses(
+        self, node: Node, parent_kind: Optional[str] = None
+    ) -> None:
+        if node.kind == self.Kind.PARENTHESES:
+            (child,) = node.children
+            if self._is_redundant_parenthesized_child(parent_kind, child.kind):
+                raise InvalidConditionExpression(
+                    "The expression has redundant parentheses;"
+                )
+            self._assert_no_redundant_parentheses(child, parent_kind)
+            return
+
+        for child in node.children:
+            self._assert_no_redundant_parentheses(child, node.kind)
+
+    def _is_redundant_parenthesized_child(
+        self, parent_kind: Optional[str], child_kind: str
+    ) -> bool:
+        if parent_kind == self.Kind.OR:
+            return child_kind in {self.Kind.OR, self.Kind.AND}
+        if parent_kind == self.Kind.AND:
+            return child_kind == self.Kind.AND
+        return False
+
     def _assert(self, condition: bool, message: str, nodes: Iterable[Node]) -> None:
         if not condition:
             raise ValueError(message + " " + " ".join([t.text for t in nodes]))

tests/test_dynamodb/test_dynamodb_condition_expressions.py

@@ -428,6 +428,89 @@ def test_condition_expression_with_reserved_keyword_as_attr_name():
     assert item == {"id": "key-0", "first": {}}
 
 
+@mock_aws
+def test_condition_expression_rejects_redundant_parentheses():
+    client = boto3.client("dynamodb", region_name="us-east-1")
+    table_name = f"T{uuid4()}"
+
+    client.create_table(
+        TableName=table_name,
+        KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}],
+        AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"}],
+        BillingMode="PAY_PER_REQUEST",
+    )
+    client.put_item(
+        TableName=table_name,
+        Item={
+            "pk": {"S": "pk"},
+            "a": {"S": "match"},
+            "c": {"S": "match"},
+            "e": {"S": "match"},
+        },
+    )
+
+    with pytest.raises(ClientError) as exc:
+        client.update_item(
+            TableName=table_name,
+            Key={"pk": {"S": "pk"}},
+            UpdateExpression="SET #z = :z",
+            ConditionExpression="#a = :b OR (#c = :d AND #e = :f)",
+            ExpressionAttributeNames={"#a": "a", "#c": "c", "#e": "e", "#z": "z"},
+            ExpressionAttributeValues={
+                ":b": {"S": "match"},
+                ":d": {"S": "match"},
+                ":f": {"S": "match"},
+                ":z": {"S": "updated"},
+            },
+        )
+
+    err = exc.value.response["Error"]
+    assert err["Code"] == "ValidationException"
+    assert (
+        err["Message"]
+        == "Invalid ConditionExpression: The expression has redundant parentheses;"
+    )
+
+
+@mock_aws
+def test_condition_expression_allows_required_parentheses():
+    client = boto3.client("dynamodb", region_name="us-east-1")
+    table_name = f"T{uuid4()}"
+
+    client.create_table(
+        TableName=table_name,
+        KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}],
+        AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"}],
+        BillingMode="PAY_PER_REQUEST",
+    )
+    client.put_item(
+        TableName=table_name,
+        Item={
+            "pk": {"S": "pk"},
+            "a": {"S": "match"},
+            "c": {"S": "match"},
+            "e": {"S": "match"},
+        },
+    )
+
+    client.update_item(
+        TableName=table_name,
+        Key={"pk": {"S": "pk"}},
+        UpdateExpression="SET #z = :z",
+        ConditionExpression="#a = :b AND (#c = :d OR #e = :f)",
+        ExpressionAttributeNames={"#a": "a", "#c": "c", "#e": "e", "#z": "z"},
+        ExpressionAttributeValues={
+            ":b": {"S": "match"},
+            ":d": {"S": "match"},
+            ":f": {"S": "nope"},
+            ":z": {"S": "updated"},
+        },
+    )
+
+    item = client.get_item(TableName=table_name, Key={"pk": {"S": "pk"}})["Item"]
+    assert item["z"] == {"S": "updated"}
+
+
 @mock_aws
 def test_condition_check_failure_exception_is_raised_when_values_are_returned_for_an_item_with_a_top_level_list():
     # This explicitly tests for a failure in handling JSONification of DynamoType