List only .zip archives as packaged Dags

airflow-core/docs/core-concepts/dags.rst

@@ -719,6 +719,8 @@ You can either do this all inside of the Dag bundle, with a standard filesystem
     package1/__init__.py
     package1/functions.py
 
+The file name must end in ``.zip``. Other zip archives in the Dag bundle, such as JAR files, are not parsed as Dags.
+
 Note that packaged Dags come with some caveats:
 
 * They cannot be used if you have pickling enabled for serialization

airflow-core/newsfragments/74033.significant.rst

@@ -0,0 +1,3 @@
+Only ``.zip`` archives are discovered as packaged Dags
+
+Dag discovery now lists a zip archive as a packaged Dag only when its file name ends in ``.zip``. Before, every zip archive in a Dag bundle was parsed whatever its name, including JAR files and archives with no suffix. A packaged Dag stored under another name is no longer found: rename it to end in ``.zip``. ``.py`` files are listed as before.

airflow-core/src/airflow/utils/file.py

@@ -110,7 +110,9 @@ def find_dag_file_paths(directory: str | os.PathLike[str], safe_mode: bool) -> l
     for file_path in find_path_from_directory(directory, ".airflowignore", ignore_file_syntax):
         path = Path(file_path)
         try:
-            if path.is_file() and (path.suffix == ".py" or zipfile.is_zipfile(path)):
+            if path.is_file() and (
+                path.suffix == ".py" or (path.suffix.lower() == ".zip" and zipfile.is_zipfile(path))
+            ):
                 if might_contain_dag(file_path, safe_mode, conf=conf):
                     file_paths.append(file_path)
         except Exception:

airflow-core/tests/unit/utils/test_file.py

@@ -213,6 +213,18 @@ def test_list_py_file_paths(self, test_zip_path):
             f"Detected files mismatched expected files:\ndetected_files: {pformat(detected_files)}\nexpected_files: {pformat(expected_files)}"
         )
 
+    @pytest.mark.parametrize("archive_name", ["task_handler.jar", "no_suffix"])
+    def test_list_py_file_paths_skips_zip_archive_without_zip_suffix(self, tmp_path, archive_name):
+        dag_source = "from airflow.sdk import DAG\n"
+        (tmp_path / "dag.py").write_text(dag_source)
+        for name in ("lower.zip", "upper.ZIP", archive_name):
+            with zipfile.ZipFile(tmp_path / name, "w") as zf:
+                zf.writestr("dag.py", dag_source)
+
+        assert set(list_py_file_paths(tmp_path)) == {
+            str(tmp_path / name) for name in ("dag.py", "lower.zip", "upper.ZIP")
+        }
+
 
 @pytest.mark.parametrize(
     ("edge_filename", "expected_modification"),