Fix DAG processor detecting non-zip archives as DAG bundles

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

@@ -107,7 +107,7 @@ 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 == ".zip" and zipfile.is_zipfile(path))):
                 if might_contain_dag(file_path, safe_mode):
                     file_paths.append(file_path)
         except Exception:

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

@@ -180,6 +180,26 @@ def test_get_modules_from_invalid_file(self):
 
         assert len(modules) == 0
 
+    @mock.patch("zipfile.is_zipfile")
+    def test_find_dag_file_paths_ignores_non_zip_archives(self, mock_is_zipfile, tmp_path):
+        mock_is_zipfile.return_value = True
+
+        (tmp_path / "test.jar").touch()
+        (tmp_path / "test.docx").touch()
+        (tmp_path / "test.zip").touch()
+        (tmp_path / "test.py").touch()
+
+        with mock.patch("airflow.utils.file.might_contain_dag") as mock_might_contain_dag:
+            mock_might_contain_dag.return_value = True
+            paths = file_utils.find_dag_file_paths(os.fspath(tmp_path), safe_mode=False)
+
+        assert len(paths) == 2
+        paths_str = " ".join(paths)
+        assert "test.zip" in paths_str
+        assert "test.py" in paths_str
+        assert "test.jar" not in paths_str
+        assert "test.docx" not in paths_str
+
     def test_list_py_file_paths(self, test_zip_path):
         detected_files = set()
         expected_files = set()