pierrejeambrun commented on code in PR #64610:
URL: https://github.com/apache/airflow/pull/64610#discussion_r3536652409
##########
airflow-core/src/airflow/api_fastapi/common/parameters.py:
##########
@@ -515,6 +516,71 @@ def depends_search(
return depends_search
+class _RegexParam(BaseParam[str]):
+ """
+ Filter using database-level regex matching (regexp_match).
+
+ SQLAlchemy's ``regexp_match`` is supported on PostgreSQL (``~`` operator),
+ MySQL/MariaDB (``REGEXP``), and SQLite (via Python ``re.match``).
+ Note: SQLite uses ``re.match`` semantics (anchored at the start of the
+ string), while PostgreSQL uses ``re.search`` (matches anywhere).
+ """
+
+ def __init__(self, attribute: ColumnElement, skip_none: bool = True) ->
None:
+ super().__init__(skip_none=skip_none)
+ self.attribute: ColumnElement = attribute
+
+ def to_orm(self, select: Select) -> Select:
+ if self.value is None and self.skip_none:
+ return select
+ return select.where(self.attribute.regexp_match(self.value))
+
+ @classmethod
+ def depends(cls, *args: Any, **kwargs: Any) -> Self:
+ raise NotImplementedError("Use regex_param_factory instead, depends is
not implemented.")
+
+
+def _validate_regex_pattern(value: str | None) -> str | None:
+ """Validate that the regex pattern is enabled and syntactically correct."""
+ if value is None:
+ return value
+ if not conf.getboolean("api", "enable_regexp_query_filters"):
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail="Regex query filters are disabled. Set [api]
enable_regexp_query_filters = True to use them.",
+ )
+ try:
+ re.compile(value)
+ except re.error as e:
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail=f"Invalid regular expression: {e}",
+ )
+ return value
+
+
+_DEFAULT_REGEX_DESCRIPTION = (
+ "Regex filter. Uses database-native regex "
+ "(PostgreSQL ~ operator, MySQL REGEXP, SQLite re.match). "
+ "Note: on SQLite, matching is anchored at the start of the string."
+)
Review Comment:
I wouldn't mention those implementation details (baackend implementation
difference) as part of the public API. I would prefer for this to remain
internals.
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/asset_events.py:
##########
@@ -44,7 +47,7 @@ def _get_asset_events_through_sql_clauses(
) -> AssetEventsResponse:
order_by_clause = AssetEvent.timestamp.asc() if ascending else
AssetEvent.timestamp.desc()
asset_events_query =
select(AssetEvent).join(join_clause).where(where_clause).order_by(order_by_clause)
- if limit:
+ if limit is not None:
Review Comment:
Is this change related ?
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py:
##########
@@ -319,12 +322,24 @@ def get_asset_events(
source_map_index: Annotated[
FilterParam[int | None],
Depends(filter_param_factory(AssetEvent.source_map_index, int | None))
],
+ partition_key: QueryAssetEventPartitionKeyFilter,
+ partition_key_pattern: QueryAssetEventPartitionKeyRegex,
name_pattern: QueryAssetNamePatternSearch,
name_prefix_pattern: QueryAssetNamePrefixPatternSearch,
timestamp_range: Annotated[RangeFilter,
Depends(datetime_range_filter_factory("timestamp", AssetEvent))],
session: SessionDep,
) -> AssetEventCollectionResponse:
"""Get asset events."""
+ if partition_key.value is not None and partition_key_pattern.value is not
None:
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail="partition_key and partition_key_pattern are mutually
exclusive.",
+ )
+
+ if partition_key_pattern.value is not None:
+ # Bound the runtime of the user-supplied regex to guard against ReDoS
(PostgreSQL only).
Review Comment:
At this point it's called for every backend.
```suggestion
# Bound the runtime of the user-supplied regex to guard against
ReDoS.
```
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/asset_events.py:
##########
@@ -120,13 +176,32 @@ def get_asset_event_by_asset_alias(
before: Annotated[UtcDateTime | None, Query(description="The end of the
time range")] = None,
ascending: Annotated[bool, Query(description="Whether to sort results in
ascending order")] = True,
limit: Annotated[int | None, Query(description="The maximum number of
results to return")] = None,
+ partition_key: Annotated[
+ str | None,
+ Query(description="Exact partition key filter."),
+ ] = None,
+ partition_key_pattern: Annotated[
+ str | None,
+ Query(
+ description="Partition key regex filter. Uses database-native
regex "
+ "(PostgreSQL ~ operator, MySQL REGEXP, SQLite re.match). "
+ "Note: on SQLite, matching is anchored at the start of the string."
+ ),
+ ] = None,
Review Comment:
Why not use `QueryAssetEventPartitionKeyRegex` and the other one?
This create duplicated validation code in the view
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -1635,6 +1635,39 @@ api:
type: boolean
example: ~
default: "True"
+ enable_regexp_query_filters:
+ description: |
+ Whether to enable regular-expression based query filters in the API.
Currently this
+ controls the ``partition_key_pattern`` filter on ``GET
/assets/events`` and on the
+ Execution API asset-event endpoints.
+
+ Disabled by default for security reasons. When enabled, a
user-supplied regular
+ expression is passed to the database's own regex engine, which is a
Regular expression
+ Denial of Service (ReDoS) vector: a crafted pattern can force the
engine into
+ pathological backtracking and consume database backend CPU. Only
enable this if you
+ accept that trade-off, and keep ``regexp_query_timeout`` set to a
sensible value.
+
+ Exact-match ``partition_key`` filtering is always available and is not
affected by this
Review Comment:
```suggestion
```
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -1635,6 +1635,39 @@ api:
type: boolean
example: ~
default: "True"
+ enable_regexp_query_filters:
+ description: |
+ Whether to enable regular-expression based query filters in the API.
Currently this
+ controls the ``partition_key_pattern`` filter on ``GET
/assets/events`` and on the
+ Execution API asset-event endpoints.
+
+ Disabled by default for security reasons. When enabled, a
user-supplied regular
+ expression is passed to the database's own regex engine, which is a
Regular expression
+ Denial of Service (ReDoS) vector: a crafted pattern can force the
engine into
+ pathological backtracking and consume database backend CPU. Only
enable this if you
+ accept that trade-off, and keep ``regexp_query_timeout`` set to a
sensible value.
+
+ Exact-match ``partition_key`` filtering is always available and is not
affected by this
+ setting.
+ version_added: 3.4.0
+ type: boolean
+ example: ~
+ default: "False"
+ regexp_query_timeout:
+ description: |
+ Maximum time, in seconds, that a query using a regular-expression
filter (see
+ ``enable_regexp_query_filters``) is allowed to run.
+
+ This is the primary runtime mitigation for the ReDoS risk described
above: on PostgreSQL
+ it is enforced as a transaction-local ``statement_timeout`` so that a
pathological
+ pattern is aborted instead of pinning a database backend (PostgreSQL
has no built-in
+ regex step limit, unlike MySQL's ``regexp_time_limit``). Keep it as
low as your
+ legitimate queries allow. A value of ``0`` disables the timeout, which
is not
+ recommended while the feature is enabled.
+ version_added: 3.4.0
Review Comment:
We don't need two settings, one setting only with `None` (deactivated), or a
number value `the timeout` it's up.
This will prevent the 'true/0' combo which is 'allow regexp matching but no
timeout enforced.
##########
airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml:
##########
@@ -433,6 +433,28 @@ paths:
- type: integer
- type: 'null'
title: Source Map Index
+ - name: partition_key
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: string
+ - type: 'null'
+ title: Partition Key
+ - name: partition_key_pattern
Review Comment:
I wouldn't call this `partition_key_pattern` this makes it look like othe
substring matching filters on the API while it's not.
I would use a new convention, maybe `partition_key_regexp_pattern` or
something to differenciate them with that new class of Regexp filters.
Otherwise people will get confused as to why `partition_key_pattern` behaves
differently than other `*_pattern`
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py:
##########
@@ -319,12 +322,24 @@ def get_asset_events(
source_map_index: Annotated[
FilterParam[int | None],
Depends(filter_param_factory(AssetEvent.source_map_index, int | None))
],
+ partition_key: QueryAssetEventPartitionKeyFilter,
+ partition_key_pattern: QueryAssetEventPartitionKeyRegex,
name_pattern: QueryAssetNamePatternSearch,
name_prefix_pattern: QueryAssetNamePrefixPatternSearch,
timestamp_range: Annotated[RangeFilter,
Depends(datetime_range_filter_factory("timestamp", AssetEvent))],
session: SessionDep,
) -> AssetEventCollectionResponse:
"""Get asset events."""
+ if partition_key.value is not None and partition_key_pattern.value is not
None:
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail="partition_key and partition_key_pattern are mutually
exclusive.",
+ )
Review Comment:
I would remove this.
Filters aren't mutually exclusive, if someone want to apply both, fine it
should yeild 0 results.
##########
airflow-core/src/airflow/utils/sqlalchemy.py:
##########
@@ -61,6 +61,26 @@ def get_dialect_name(session: Session) -> str | None:
return getattr(bind.dialect, "name", None)
+def apply_regex_query_timeout(session: Session) -> None:
+ """
+ Bound the runtime of a user-supplied regex filter on PostgreSQL.
+
+ Reads the ``[api] regexp_query_timeout`` config (in seconds) and sets a
transaction-local
+ ``statement_timeout`` (via ``set_config(..., is_local=True)``) so it
applies only to the current
+ transaction and resets automatically at commit/rollback. This is a ReDoS
safeguard: a malicious
+ pattern is aborted instead of pinning a database backend. No-op on
non-PostgreSQL backends and
+ when the configured timeout is ``0`` (disabled).
+ """
+ timeout_seconds = conf.getint("api", "regexp_query_timeout")
+ if timeout_seconds <= 0:
+ return
+ if get_dialect_name(session) == "postgresql":
+ timeout_ms = timeout_seconds * 1000
+ session.execute(
+ text("SELECT set_config('statement_timeout', :timeout,
true)").bindparams(timeout=str(timeout_ms))
+ )
+
Review Comment:
Should this be move to the `to_orm` method of the regexp filter? This way
we're sure it's never 'forgotten' when running a regexp filter query?
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -1635,6 +1635,39 @@ api:
type: boolean
example: ~
default: "True"
+ enable_regexp_query_filters:
+ description: |
+ Whether to enable regular-expression based query filters in the API.
Currently this
+ controls the ``partition_key_pattern`` filter on ``GET
/assets/events`` and on the
+ Execution API asset-event endpoints.
+
+ Disabled by default for security reasons. When enabled, a
user-supplied regular
+ expression is passed to the database's own regex engine, which is a
Regular expression
+ Denial of Service (ReDoS) vector: a crafted pattern can force the
engine into
+ pathological backtracking and consume database backend CPU. Only
enable this if you
+ accept that trade-off, and keep ``regexp_query_timeout`` set to a
sensible value.
Review Comment:
"When enabled allows additional filtering feature that requires db side
regexp matching..."
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -858,16 +858,24 @@ def get(
before: datetime | None = None,
ascending: bool = True,
limit: int | None = None,
+ partition_key: str | None = None,
+ partition_key_pattern: str | None = None,
) -> AssetEventsResponse:
"""Get Asset event from the API server."""
+ if partition_key is not None and partition_key_pattern is not None:
+ raise ValueError("partition_key and partition_key_pattern are
mutually exclusive")
common_params: dict[str, Any] = {}
if after:
common_params["after"] = after.isoformat()
if before:
common_params["before"] = before.isoformat()
common_params["ascending"] = ascending
- if limit:
+ if limit is not None:
common_params["limit"] = limit
+ if partition_key is not None:
+ common_params["partition_key"] = partition_key
+ if partition_key_pattern is not None:
+ common_params["partition_key_pattern"] = partition_key_pattern
Review Comment:
'is not None' Is that necessary? ("" partitions should sill go under this?)
and `0` limit too ?
##########
airflow-core/src/airflow/utils/sqlalchemy.py:
##########
@@ -61,6 +61,26 @@ def get_dialect_name(session: Session) -> str | None:
return getattr(bind.dialect, "name", None)
+def apply_regex_query_timeout(session: Session) -> None:
+ """
+ Bound the runtime of a user-supplied regex filter on PostgreSQL.
+
+ Reads the ``[api] regexp_query_timeout`` config (in seconds) and sets a
transaction-local
+ ``statement_timeout`` (via ``set_config(..., is_local=True)``) so it
applies only to the current
+ transaction and resets automatically at commit/rollback. This is a ReDoS
safeguard: a malicious
+ pattern is aborted instead of pinning a database backend. No-op on
non-PostgreSQL backends and
+ when the configured timeout is ``0`` (disabled).
+ """
+ timeout_seconds = conf.getint("api", "regexp_query_timeout")
+ if timeout_seconds <= 0:
+ return
+ if get_dialect_name(session) == "postgresql":
+ timeout_ms = timeout_seconds * 1000
+ session.execute(
+ text("SELECT set_config('statement_timeout', :timeout,
true)").bindparams(timeout=str(timeout_ms))
+ )
+
Review Comment:
Also there is a side effect for all statements running after this, in the
same transaction (all other db queries for the entire request basically)
##########
airflow-core/src/airflow/api_fastapi/common/parameters.py:
##########
@@ -515,6 +516,71 @@ def depends_search(
return depends_search
+class _RegexParam(BaseParam[str]):
+ """
+ Filter using database-level regex matching (regexp_match).
+
+ SQLAlchemy's ``regexp_match`` is supported on PostgreSQL (``~`` operator),
+ MySQL/MariaDB (``REGEXP``), and SQLite (via Python ``re.match``).
+ Note: SQLite uses ``re.match`` semantics (anchored at the start of the
+ string), while PostgreSQL uses ``re.search`` (matches anywhere).
+ """
+
+ def __init__(self, attribute: ColumnElement, skip_none: bool = True) ->
None:
+ super().__init__(skip_none=skip_none)
+ self.attribute: ColumnElement = attribute
+
+ def to_orm(self, select: Select) -> Select:
+ if self.value is None and self.skip_none:
+ return select
+ return select.where(self.attribute.regexp_match(self.value))
+
+ @classmethod
+ def depends(cls, *args: Any, **kwargs: Any) -> Self:
+ raise NotImplementedError("Use regex_param_factory instead, depends is
not implemented.")
+
+
+def _validate_regex_pattern(value: str | None) -> str | None:
+ """Validate that the regex pattern is enabled and syntactically correct."""
+ if value is None:
+ return value
+ if not conf.getboolean("api", "enable_regexp_query_filters"):
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail="Regex query filters are disabled. Set [api]
enable_regexp_query_filters = True to use them.",
+ )
+ try:
+ re.compile(value)
+ except re.error as e:
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail=f"Invalid regular expression: {e}",
+ )
+ return value
Review Comment:
Move this piece inside the `depends_regex`
##########
airflow-core/src/airflow/api_fastapi/common/parameters.py:
##########
@@ -515,6 +516,71 @@ def depends_search(
return depends_search
+class _RegexParam(BaseParam[str]):
+ """
+ Filter using database-level regex matching (regexp_match).
+
+ SQLAlchemy's ``regexp_match`` is supported on PostgreSQL (``~`` operator),
+ MySQL/MariaDB (``REGEXP``), and SQLite (via Python ``re.match``).
+ Note: SQLite uses ``re.match`` semantics (anchored at the start of the
+ string), while PostgreSQL uses ``re.search`` (matches anywhere).
+ """
+
+ def __init__(self, attribute: ColumnElement, skip_none: bool = True) ->
None:
+ super().__init__(skip_none=skip_none)
+ self.attribute: ColumnElement = attribute
+
+ def to_orm(self, select: Select) -> Select:
+ if self.value is None and self.skip_none:
+ return select
+ return select.where(self.attribute.regexp_match(self.value))
+
+ @classmethod
+ def depends(cls, *args: Any, **kwargs: Any) -> Self:
+ raise NotImplementedError("Use regex_param_factory instead, depends is
not implemented.")
+
+
+def _validate_regex_pattern(value: str | None) -> str | None:
+ """Validate that the regex pattern is enabled and syntactically correct."""
+ if value is None:
+ return value
+ if not conf.getboolean("api", "enable_regexp_query_filters"):
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail="Regex query filters are disabled. Set [api]
enable_regexp_query_filters = True to use them.",
+ )
Review Comment:
This piece should be moved inside the `_RegexParam` constructor.-> 400 if
the param is provided and the setting is 'off'.
It should never be possible to instanciate the regexp filter without
checking the setting is 'allowed' beforehand.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
