ReDoS in SparkSubmitHook._mask_cmd() via user-controlled application_args (apache-airflow-providers-apache-spark)

# ReDoS in `_mask_cmd()` via User-Controlled `application_args`

## Summary
The `apache-airflow-providers-apache-spark` package contains a Regular Expression Denial of Service (ReDoS) vulnerability in the `SparkSubmitHook._mask_cmd()` method. User-controlled values injected through the Airflow REST API DAG trigger `conf` parameter are incorporated into a shell command string that is processed by a regex with catastrophic backtracking characteristics, enabling a remote DoS attack.

## Affected Package
- Package: apache-airflow-providers-apache-spark
- PyPI: https://pypi.org/project/apache-airflow-providers-apache-spark/
- Tested version: 4.10.0
- CVSS v3.1: 7.5
- Severity: High

## Vulnerability Details
In `airflow/providers/apache/spark/hooks/spark_submit.py` at lines 508–530, the `_mask_cmd()` method applies the following regex to `' '.join(connection_cmd)`:

```python
re.sub(
r'(\S*?(?:secret|password)\S*?(?:=|\s+)([\'"]?))(?:(?!\2\s).)*',
r'\1********',
' '.join(connection_cmd)
)
```

`connection_cmd` includes `self._application_args`, which is a templated field populated from the `application_args` key in the DAG trigger `conf` dict. The nested lookahead `(?:(?!\2\s).)*` combined with the `\S*?` quantifier causes **quadratic backtracking** when the input string is long and does not contain the expected pattern.

### Timing Evidence
| Input size | Time elapsed |
|-----------|-------------|
| 10,000 chars | ~2 seconds |
| 50,000 chars | ~57 seconds |

The growth rate is O(n²), consistent with quadratic backtracking.

## Attack Vector
An authenticated Airflow user with DAG trigger permissions can invoke the following REST API call:

```http
POST /api/v1/dags/{dag_id}/dagRuns
Content-Type: application/json

{
"conf": {
"application_args": ["aaaa...aaaa!"]
}
}
```

Where `aaaa...aaaa!` is a string of ~50,000 characters not containing `secret` or `password`. This causes the Airflow worker process executing the SparkSubmitHook to spin for ~57 seconds per task execution, effectively blocking the worker slot.

## Proof of Concept
```python
import re, time

regex = r'(\S*?(?:secret|password)\S*?(?:=|\s+)([\'"]?))(?:(?!\2\s).)*'

for n in [10000, 30000, 50000]:
payload = 'a' * n + '!'
t = time.time()
re.sub(regex, r'\1********', payload)
print(f"n={n}: {time.time()-t:.2f}s")
# n=10000: ~2.1s
# n=30000: ~19s
# n=50000: ~57s
```

- A remote authenticated attacker (Airflow user with trigger permission) can block Airflow worker slots indefinitely by triggering DAG runs with crafted `application_args`
- Repeated triggering can exhaust all available workers, causing a complete Denial of Service for the Airflow cluster
- No code execution or data exfiltration is possible through this vector alone

## Remediation
Replace the vulnerable regex with a possessive quantifier or atomic group equivalent, or rewrite the masking logic without a nested lookahead:

```python
# Option 1: Use re.sub with a non-backtracking approach
import re

def _mask_cmd(self, connection_cmd):
cmd_str = ' '.join(connection_cmd)
# Replace password/secret values using a non-catastrophic pattern
masked = re.sub(
r'(\S*(?:secret|password)\S*(?:=|\s+)[\'"]?)(\S+)',
r'\1********',
cmd_str
)
return masked
```

The key fix is removing the nested lookahead `(?:(?!\2\s).)*` and replacing it with a simple `\S+` or bounded quantifier that cannot exhibit catastrophic backtracking.

Read airflow/providers/apache/spark/hooks/spark_submit.py, especially SparkSubmitHook._mask_cmd(), and run the supplied Python regex timing PoC to reproduce the scaling. Trace how application_args enters connection_cmd; done means secret/password masking still works while long non-matching input no longer exhibits quadratic backtracking.

Written by the indexing model from the issue text.