Add OSS-Fuzz Atheris fuzzers for core serialization

pyproject.toml

@@ -1504,6 +1504,7 @@ members = [
     "shared/secrets_backend",
     "shared/secrets_masker",
     "shared/timezones",
+    "scripts/ossfuzz",
     # Automatically generated provider workspace members (update_airflow_pyproject_toml.py)
     "providers/airbyte",
     "providers/alibaba",

scripts/ossfuzz/README.md

@@ -0,0 +1,72 @@
+# Airflow OSS-Fuzz fuzzers
+
+This directory contains the upstream-owned fuzz targets used by OSS-Fuzz for
+Apache Airflow.
+
+## Security Model Alignment
+
+These fuzzers target code paths with **clear security boundaries** per
+Airflow's [security model](../../airflow-core/docs/security/security_model.rst):
+
+- **DAG Serialization/Deserialization**: Used by Scheduler and API Server with
+  schema validation. Input comes from DAG parsing and caching.
+- **Connection URI Parsing**: Used when creating/updating connections via API.
+
+We explicitly **avoid** fuzzing code paths in the "DAG author trust zone"
+where Airflow's policy is that DAG authors can execute arbitrary code.
+
+## What's here
+
+- `src/ossfuzz/*_fuzz.py`: Atheris fuzz targets.
+- `*.dict`: Optional libFuzzer dictionaries for structured inputs.
+- `*.options`: libFuzzer options (e.g. `max_len`) tuned per target.
+- `seed_corpus/<fuzzer>/...`: Small seed corpora that get zipped and uploaded to
+  OSS-Fuzz for each target.
+
+## Fuzzers
+
+| Fuzzer | Target | Security Boundary |
+|--------|--------|-------------------|
+| `serialized_dag_fuzz` | `DagSerialization.from_dict()` | Schema validation |
+| `connection_uri_fuzz` | `Connection._parse_from_uri()` | API input validation |
+
+## Supported engines / sanitizers (Python constraints)
+
+Airflow is fuzzed as a **Python** OSS-Fuzz project. Practically, this means:
+
+- **Fuzzing engine**: `libfuzzer` (Atheris). Other engines (AFL/honggfuzz) are
+  not typically used/supported for Python targets in OSS-Fuzz.
+- **Sanitizers**: `address`, `undefined`, `coverage`, `introspector` are the
+  relevant modes. **MSan (`memory`) is not supported** for Python OSS-Fuzz
+  projects.
+
+## Running locally
+
+From the repository root, use `uv run` which will automatically set up the
+virtual environment and install dependencies:
+
+```bash
+# Run serialized_dag_fuzz (quick test, 60 seconds)
+uv run --package apache-airflow-ossfuzz serialized_dag_fuzz -max_total_time=60
+
+# Run connection_uri_fuzz
+uv run --package apache-airflow-ossfuzz connection_uri_fuzz -max_total_time=60
+
+# Run with a corpus directory
+mkdir -p corpus/serialized_dag_fuzz
+uv run --package apache-airflow-ossfuzz serialized_dag_fuzz corpus/serialized_dag_fuzz -max_total_time=300
+```
+
+## Running with OSS-Fuzz helper
+
+From a checkout of `google/oss-fuzz`:
+
+```bash
+# Build + basic validation:
+python3 infra/helper.py build_fuzzers --clean --sanitizer address airflow /path/to/airflow
+python3 infra/helper.py check_build --sanitizer address airflow
+
+# Coverage build + validation:
+python3 infra/helper.py build_fuzzers --clean --sanitizer coverage airflow /path/to/airflow
+python3 infra/helper.py check_build --sanitizer coverage airflow
+```

scripts/ossfuzz/connection_uri_fuzz.dict

@@ -0,0 +1,19 @@
+"postgres"
+"postgresql"
+"mysql"
+"mssql"
+"sqlite"
+"http"
+"https"
+"ssh"
+"s3"
+"gcs"
+"wasb"
+"://"
+"@"
+":"
+"/"
+"?"
+"&"
+"="
+"__extra__"

scripts/ossfuzz/connection_uri_fuzz.options

@@ -0,0 +1,2 @@
+[libfuzzer]
+max_len = 2048

scripts/ossfuzz/pyproject.toml

@@ -0,0 +1,56 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[project]
+name = "apache-airflow-ossfuzz"
+description = "OSS-Fuzz harnesses for Apache Airflow"
+version = "0.0"
+readme = "README.md"
+classifiers = [
+    "Private :: Do Not Upload",
+]
+
+dependencies = [
+    "apache-airflow-core",
+    "apache-airflow-providers-standard",
+    "atheris>=2.3.0; python_version >= '3.11'",
+    "pendulum>=3.0.0",
+]
+
+[project.scripts]
+serialized_dag_fuzz = "ossfuzz.serialized_dag_fuzz:main"
+connection_uri_fuzz = "ossfuzz.connection_uri_fuzz:main"
+
+[dependency-groups]
+dev = [
+    "apache-airflow-devel-common",
+]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/ossfuzz"]
+
+[tool.ruff]
+extend = "../../pyproject.toml"
+src = ["src"]
+
+[tool.ruff.lint.per-file-ignores]
+# Ignore Doc rules et al for anything outside of tests
+"!src/*" = ["D", "S101", "TRY002"]

scripts/ossfuzz/seed_corpus/connection_uri_fuzz/http_uri.txt

@@ -0,0 +1 @@
+http://api.example.com/v1?timeout=30

scripts/ossfuzz/seed_corpus/connection_uri_fuzz/postgres_uri.txt

@@ -0,0 +1 @@
+postgres://user:password@localhost:5432/mydb

scripts/ossfuzz/seed_corpus/connection_uri_fuzz/with_extra.txt

@@ -0,0 +1 @@
+mysql://root:secret@db.local:3306/app?__extra__={"ssl": true}

scripts/ossfuzz/seed_corpus/serialized_dag_fuzz/minimal.json

@@ -0,0 +1 @@
+{"__version": 3, "dag": {"dag_id": "test_dag", "timezone": "UTC", "schedule": "@daily", "start_date": "2020-01-01T00:00:00+00:00"}}

scripts/ossfuzz/serialized_dag_fuzz.dict

@@ -0,0 +1,14 @@
+"dag_id"
+"task_id"
+"schedule"
+"start_date"
+"end_date"
+"catchup"
+"description"
+"default_args"
+"params"
+"tasks"
+"__type"
+"__var"
+"@daily"
+"@hourly"

scripts/ossfuzz/serialized_dag_fuzz.options

@@ -0,0 +1,2 @@
+[libfuzzer]
+max_len = 8192

scripts/ossfuzz/src/ossfuzz/connection_uri_fuzz.py

@@ -0,0 +1,130 @@
+#!/usr/bin/python3
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Fuzzer for Airflow Connection URI parsing.
+
+This targets the Connection._parse_from_uri() method which is used when
+creating connections via the API. Input comes from authenticated API users.
+
+Target: airflow.models.connection.Connection (uri parameter)
+Security boundary: API input validation
+"""
+
+import os
+import sys
+from urllib.parse import urlencode
+
+import atheris
+
+os.environ.setdefault("AIRFLOW_HOME", "/tmp/airflow")
+
+_CONN_TYPES = [
+    "postgres",
+    "postgresql",
+    "mysql",
+    "mssql",
+    "sqlite",
+    "http",
+    "https",
+    "ssh",
+    "s3",
+    "gcs",
+    "wasb",
+]
+_HOST_PROTOCOLS = ["http", "https", "tcp", "udp"]
+
+
+def _maybe_consume_str(fdp: atheris.FuzzedDataProvider, max_len: int) -> str | None:
+    if not fdp.ConsumeBool():
+        return None
+    return fdp.ConsumeString(max_len)
+
+
+def _build_query(fdp: atheris.FuzzedDataProvider) -> str:
+    params: dict[str, str] = {}
+    for _ in range(fdp.ConsumeIntInRange(0, 8)):
+        k = fdp.ConsumeString(16)
+        v = fdp.ConsumeString(64)
+        if k:
+            params[k] = v
+    if not params:
+        return ""
+    return "?" + urlencode(params, doseq=False)
+
+
+def _build_uri(fdp: atheris.FuzzedDataProvider) -> str:
+    conn_type = fdp.PickValueInList(_CONN_TYPES)
+    host_protocol = fdp.PickValueInList(_HOST_PROTOCOLS) if fdp.ConsumeBool() else None
+
+    login = _maybe_consume_str(fdp, 32) or ""
+    password = _maybe_consume_str(fdp, 32) or ""
+    host = _maybe_consume_str(fdp, 64) or ""
+    schema = _maybe_consume_str(fdp, 64) or ""
+    port = fdp.ConsumeIntInRange(0, 65535) if fdp.ConsumeBool() else None
+
+    authority = ""
+    if login or password:
+        authority = login
+        if password or fdp.ConsumeBool():
+            authority += ":" + password
+        authority += "@"
+
+    host_block = host
+    if port is not None and (host_block or authority):
+        host_block += f":{port}"
+    if schema:
+        host_block += f"/{schema}"
+
+    uri = f"{conn_type}://"
+    if host_protocol:
+        uri += f"{host_protocol}://"
+    uri += authority + host_block
+    uri += _build_query(fdp)
+    return uri
+
+
+with atheris.instrument_imports(include=["airflow"], enable_loader_override=False):
+    from airflow.exceptions import AirflowException
+    from airflow.models.connection import Connection, sanitize_conn_id
+
+
+def TestInput(input_bytes: bytes):
+    fdp = atheris.FuzzedDataProvider(input_bytes)
+
+    # Test sanitize_conn_id with fuzzed input
+    conn_id = fdp.ConsumeString(256)
+    _ = sanitize_conn_id(conn_id)
+
+    # Test Connection URI parsing
+    uri = _build_uri(fdp)
+    try:
+        conn = Connection(conn_id=conn_id, uri=uri)
+        _ = conn.get_uri()
+        _ = conn.extra_dejson
+    except (AirflowException, ValueError, TypeError, UnicodeError):
+        return
+
+
+def main():
+    atheris.Setup(sys.argv, TestInput, enable_python_coverage=True)
+    atheris.Fuzz()
+
+
+if __name__ == "__main__":
+    main()