Allow Viewing DagRuns and TIs if a user has DAG "read" perms (#20663)

This was updated in Airflow 2.2.0 via #16634 which restricts a user to even views the DagRuns and TI records if they don't have "edit" permissions on DAG even though it has "read" permissions.

The Behaviour seems inconsistent as a User can still view those from the Graph and Tree View of the DAG.

And since we have got `@action_has_dag_edit_access` on all the Actions like Delete/Clear etc the approach in this PR is better as when a user will try to perform any actions from the List Dag Run view like deleting the record it will give an Access Denied error.

(cherry picked from commit 05b9f3d)

airflow/www/views.py

@@ -3144,14 +3144,6 @@ def apply(self, query, func):
         return query.filter(self.model.dag_id.in_(filter_dag_ids))
 
 
-class DagEditFilter(BaseFilter):
-    """Filter using DagIDs"""
-
-    def apply(self, query, func):  # pylint: disable=redefined-outer-name,unused-argument
-        filter_dag_ids = current_app.appbuilder.sm.get_editable_dag_ids(g.user)
-        return query.filter(self.model.dag_id.in_(filter_dag_ids))
-
-
 class AirflowModelView(ModelView):
     """Airflow Mode View."""
 
@@ -3951,7 +3943,7 @@ class DagRunModelView(AirflowPrivilegeVerifierModelView):
 
     base_order = ('execution_date', 'desc')
 
-    base_filters = [['dag_id', DagEditFilter, lambda: []]]
+    base_filters = [['dag_id', DagFilter, lambda: []]]
 
     edit_form = DagRunEditForm
 
@@ -4299,7 +4291,7 @@ class TaskInstanceModelView(AirflowPrivilegeVerifierModelView):
 
     base_order = ('job_id', 'asc')
 
-    base_filters = [['dag_id', DagEditFilter, lambda: []]]
+    base_filters = [['dag_id', DagFilter, lambda: []]]
 
     def log_url_formatter(self):
         """Formats log URL."""

tests/www/views/test_views_dagrun.py

@@ -15,6 +15,8 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
+import flask
+import markupsafe
 import pytest
 import werkzeug
 
@@ -73,6 +75,21 @@ def reset_dagrun():
         session.query(TaskInstance).delete()
 
 
+def test_get_dagrun_can_view_dags_without_edit_perms(session, running_dag_run, client_dr_without_dag_edit):
+    """Test that a user without dag_edit but with dag_read permission can view the records"""
+    assert session.query(DagRun).filter(DagRun.dag_id == running_dag_run.dag_id).count() == 1
+    resp = client_dr_without_dag_edit.get('/dagrun/list/', follow_redirects=True)
+
+    with client_dr_without_dag_edit.application.test_request_context():
+        url = flask.url_for(
+            'Airflow.graph', dag_id=running_dag_run.dag_id, execution_date=running_dag_run.execution_date
+        )
+        dag_url_link = markupsafe.Markup('<a href="{url}">{dag_id}</a>').format(
+            url=url, dag_id=running_dag_run.dag_id
+        )
+    check_content_in_response(dag_url_link, resp)
+
+
 def test_create_dagrun_permission_denied(session, client_dr_without_dag_edit):
     data = {
         "state": "running",
@@ -102,7 +119,7 @@ def running_dag_run(session):
         TaskInstance(dag.get_task("runme_1"), run_id=dr.run_id, state="failed"),
     ]
     session.bulk_save_objects(tis)
-    session.flush()
+    session.commit()
     return dr
 
 
@@ -113,12 +130,12 @@ def test_delete_dagrun(session, admin_client, running_dag_run):
     assert session.query(DagRun).filter(DagRun.dag_id == running_dag_run.dag_id).count() == 0
 
 
-def test_delete_dagrun_permission_denied(session, client_dr_without_dag_edit, running_dag_run):
+def test_delete_dagrun_permission_denied(session, running_dag_run, client_dr_without_dag_edit):
     composite_key = _get_appbuilder_pk_string(DagRunModelView, running_dag_run)
 
     assert session.query(DagRun).filter(DagRun.dag_id == running_dag_run.dag_id).count() == 1
     resp = client_dr_without_dag_edit.post(f"/dagrun/delete/{composite_key}", follow_redirects=True)
-    assert resp.status_code == 404  # If it doesn't fully succeed it gives a 404.
+    check_content_in_response(f"Access denied for dag_id {running_dag_run.dag_id}", resp)
     assert session.query(DagRun).filter(DagRun.dag_id == running_dag_run.dag_id).count() == 1
 
 

tests/www/views/test_views_tasks.py

@@ -635,13 +635,15 @@ def test_task_instance_delete_permission_denied(session, client_ti_without_dag_e
         task_id="test_task_instance_delete_permission_denied",
         execution_date=timezone.utcnow(),
         state=State.DEFERRED,
+        session=session,
     )
+    session.commit()
     composite_key = _get_appbuilder_pk_string(TaskInstanceModelView, task_instance_to_delete)
     task_id = task_instance_to_delete.task_id
 
     assert session.query(TaskInstance).filter(TaskInstance.task_id == task_id).count() == 1
     resp = client_ti_without_dag_edit.post(f"/taskinstance/delete/{composite_key}", follow_redirects=True)
-    assert resp.status_code == 404  # If it doesn't fully succeed it gives a 404.
+    check_content_in_response(f"Access denied for dag_id {task_instance_to_delete.dag_id}", resp)
     assert session.query(TaskInstance).filter(TaskInstance.task_id == task_id).count() == 1