diff --git a/task-sdk/src/airflow/sdk/definitions/_internal/node.py b/task-sdk/src/airflow/sdk/definitions/_internal/node.py index 803ca837825fc..77eae83550e47 100644 --- a/task-sdk/src/airflow/sdk/definitions/_internal/node.py +++ b/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): diff --git a/task-sdk/tests/task_sdk/bases/test_operator.py b/task-sdk/tests/task_sdk/bases/test_operator.py index dcb5240a83dc8..1a653ed112b82 100644 --- a/task-sdk/tests/task_sdk/bases/test_operator.py +++ b/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) diff --git a/task-sdk/tests/task_sdk/definitions/test_dag.py b/task-sdk/tests/task_sdk/definitions/test_dag.py index 9b76816886c76..65f7306e92e06 100644 --- a/task-sdk/tests/task_sdk/definitions/test_dag.py +++ b/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,6 +69,12 @@ 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): @@ -74,6 +82,10 @@ def test_dag_id_validation(self, dag_id, exc_type, exc_value): 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"})