Reject consecutive dots in Dag and task IDs in the Task SDK

task-sdk/src/airflow/sdk/definitions/_internal/node.py

@@ -24,6 +24,7 @@
 from typing import TYPE_CHECKING, Any
 
 from airflow.sdk._shared.dagnode.node import GenericDAGNode
+from airflow.sdk.configuration import conf
 from airflow.sdk.definitions._internal.mixins import DependencyMixin
 
 if TYPE_CHECKING:
@@ -50,6 +51,8 @@ def validate_key(k: str, max_length: int = 250):
             f"The key {k!r} has to be made of alphanumeric characters, dashes, "
             f"dots, and underscores exclusively"
         )
+    if ".." in k and not conf.getboolean("core", "allow_double_dot_in_ids", fallback=False):
+        raise ValueError(f"The key {k!r} must not contain consecutive dots ('..') to prevent path traversal")
 
 
 def validate_group_key(k: str, max_length: int = 200):

task-sdk/tests/task_sdk/bases/test_operator.py

@@ -140,6 +140,14 @@ def test_baseoperator_raises_exception_when_task_id_plus_taskgroup_id_exceeds_25
             with pytest.raises(ValueError, match="The key has to be less than 250 characters"):
                 BaseOperator(task_id="1" * 249)
 
+    def test_baseoperator_rejects_task_id_with_consecutive_dots(self):
+        with DAG(dag_id="foo"):
+            with pytest.raises(
+                ValueError,
+                match=r"The key 'a\.\.b' must not contain consecutive dots \('\.\.'\) to prevent path traversal",
+            ):
+                BaseOperator(task_id="a..b")
+
     def test_baseoperator_with_task_id_and_taskgroup_id_less_than_250_chars(self):
         with DAG(dag_id="foo", schedule=None), TaskGroup("A" * 10):
             BaseOperator(task_id="1" * 239)

task-sdk/tests/task_sdk/definitions/test_dag.py

@@ -41,6 +41,8 @@
 from airflow.sdk.exceptions import AirflowDagCycleException, DuplicateTaskIdFound, RemovedInAirflow4Warning
 from airflow.utils.types import DagRunType
 
+from tests_common.test_utils.config import conf_vars
+
 DEFAULT_DATE = datetime(2016, 1, 1, tzinfo=timezone.utc)
 
 
@@ -67,13 +69,23 @@ class TestDag:
                 "dots, and underscores exclusively",
                 id="illegal",
             ),
+            pytest.param(
+                "a..b",
+                ValueError,
+                "The key 'a..b' must not contain consecutive dots ('..') to prevent path traversal",
+                id="double-dot",
+            ),
         ],
     )
     def test_dag_id_validation(self, dag_id, exc_type, exc_value):
         with pytest.raises(exc_type) as ctx:
             DAG(dag_id)
         assert str(ctx.value) == exc_value
 
+    @conf_vars({("core", "allow_double_dot_in_ids"): "True"})
+    def test_dag_id_allows_double_dots_when_enabled(self):
+        assert DAG("a..b").dag_id == "a..b"
+
     def test_dag_topological_sort_dag_without_tasks(self):
         dag = DAG("dag", schedule=None, start_date=DEFAULT_DATE, default_args={"owner": "owner1"})