[AIRFLOW-836] Use POST and CSRF for state changing endpoints

Closes #2054 from saguziel/aguziel-use-post

Author: saguziel

airflow/www/templates/admin/master.html

@@ -38,6 +38,14 @@
         alert('{{ hostname }}');
     });
     $('span').tooltip();
+
+    $.ajaxSetup({
+      beforeSend: function(xhr, settings) {
+        if (!/^(GET|HEAD|OPTIONS|TRACE)$/i.test(settings.type) && !this.crossDomain) {
+          xhr.setRequestHeader("X-CSRFToken", "{{ csrf_token() }}");
+        }
+      }
+    });
 </script>
 {% endblock %}
 

airflow/www/templates/airflow/dag.html

@@ -32,7 +32,7 @@ <h3 class="pull-left">
       {% if dag.parent_dag %}
         <span style='color:#AAA;'>SUBDAG: </span> <span> {{ dag.dag_id }}</span>
       {% else %}
-        <input id="pause_resume" dag_id="{{ dag.dag_id }}" type="checkbox" {{ "checked" if not dag.is_paused else "" }} data-toggle="toggle" data-size="mini">
+        <input id="pause_resume" dag_id="{{ dag.dag_id }}" type="checkbox" {{ "checked" if not dag.is_paused else "" }} data-toggle="toggle" data-size="mini" method="post">
         <span style='color:#AAA;'>DAG: </span> <span> {{ dag.dag_id }}</span> <small class="text-muted"> {{ dag.description }} </small>
       {% endif %}
       {% if root %}
@@ -364,7 +364,7 @@ <h4 class="modal-title" id="myModalLabel">
         is_paused = 'false'
       }
       url = "{{ url_for('airflow.paused') }}" + '?is_paused=' + is_paused + '&dag_id=' + dag_id;
-      $.ajax(url);
+      $.post(url);
     });
 
   </script>

airflow/www/templates/airflow/dags.html

@@ -66,7 +66,7 @@ <h2>DAGs</h2>
                 <!-- Column 2: Turn dag on/off -->
                 <td>
                   {% if dag_id in orm_dags %}
-                    <input id="toggle-{{ dag_id }}" dag_id="{{ dag_id }}" type="checkbox" {{ "checked" if not orm_dags[dag_id].is_paused else "" }} data-toggle="toggle" data-size="mini">
+                    <input id="toggle-{{ dag_id }}" dag_id="{{ dag_id }}" type="checkbox" {{ "checked" if not orm_dags[dag_id].is_paused else "" }} data-toggle="toggle" data-size="mini" method="post">
                   {% endif %}
                 </td>
 
@@ -214,7 +214,7 @@ <h2>DAGs</h2>
             is_paused = 'false'
           }
           url = 'airflow/paused?is_paused=' + is_paused + '&dag_id=' + dag_id;
-          $.ajax(url);
+          $.post(url);
         });
       });
       $('#dags').dataTable({

airflow/www/templates/airflow/query.html

@@ -29,12 +29,12 @@
 
 {% block body %}
   <h2>Ad Hoc Query</h2>
-  <form method="get" id="query_form">
+  <form method="post" id="query_form">
     <div class="form-inline">
         <input name="_csrf_token" type="hidden" value="{{ csrf_token() }}">
         {{ form.conn_id }}
-        <input type="submit" class="btn btn-default" value="Run!">
-        <input type="button" class="btn btn-default" value=".csv" id="csv">
+        <input type="submit" class="btn btn-default" value="Run!" id="submit_without_csv">
+        <input type="submit" class="btn btn-default" value=".csv" id="submit_with_csv">
         <span id="results"></span><br>
         <div id='ace_container'>
             {{ form.sql(rows=10) }}
@@ -71,12 +71,16 @@ <h2>Ad Hoc Query</h2>
         });
         $('select').addClass("form-control");
         sync();
-        $("#query_form").submit(function(event){
+        $("#submit_without_csv").submit(function(event){
             $("#results").html("<img width='25'src='{{ url_for('static', filename='loading.gif') }}'>");
         });
-        $("#csv").on("click", function(){
-            window.location += '&csv=true';
-        })
+        $("#submit_with_csv").click(function(){
+          $("#csv_value").remove();
+          $("#query_form").append('<input name="csv" type="hidden" value="true" id="csv_value">');
+        });
+        $("#submit_without_csv").click(function(){
+          $("#csv_value").remove();
+        });
     });
   </script>
 {% endblock %}

airflow/www/views.py

@@ -1597,7 +1597,7 @@ def landing_times(self):
             form=form,
         )
 
-    @expose('/paused')
+    @expose('/paused', methods=['POST'])
     @login_required
     @wwwutils.action_logging
     def paused(self):
@@ -1865,7 +1865,7 @@ def index(self):
 
 
 class QueryView(wwwutils.DataProfilingMixin, BaseView):
-    @expose('/')
+    @expose('/', methods=['POST', 'GET'])
     @wwwutils.gzipped
     def query(self):
         session = settings.Session()
@@ -1874,9 +1874,9 @@ def query(self):
         session.expunge_all()
         db_choices = list(
             ((db.conn_id, db.conn_id) for db in dbs if db.get_hook()))
-        conn_id_str = request.args.get('conn_id')
-        csv = request.args.get('csv') == "true"
-        sql = request.args.get('sql')
+        conn_id_str = request.form.get('conn_id')
+        csv = request.form.get('csv') == "true"
+        sql = request.form.get('sql')
 
         class QueryForm(Form):
             conn_id = SelectField("Layout", choices=db_choices)

tests/core.py

@@ -1407,6 +1407,44 @@ def test_variables(self):
         os.remove('variables1.json')
         os.remove('variables2.json')
 
+class CSRFTests(unittest.TestCase):
+    def setUp(self):
+        configuration.load_test_config()
+        configuration.conf.set("webserver", "authenticate", "False")
+        configuration.conf.set("webserver", "expose_config", "True")
+        app = application.create_app()
+        app.config['TESTING'] = True
+        self.app = app.test_client()
+
+        self.dagbag = models.DagBag(
+            dag_folder=DEV_NULL, include_examples=True)
+        self.dag_bash = self.dagbag.dags['example_bash_operator']
+        self.runme_0 = self.dag_bash.get_task('runme_0')
+
+    def get_csrf(self, response):
+        tree = html.fromstring(response.data)
+        form = tree.find('.//form')
+
+        return form.find('.//input[@name="_csrf_token"]').value
+
+    def test_csrf_rejection(self):
+        endpoints = ([
+            "/admin/queryview/",
+            "/admin/airflow/paused?dag_id=example_python_operator&is_paused=false",
+        ])
+        for endpoint in endpoints:
+            response = self.app.post(endpoint)
+            self.assertIn('CSRF token is missing', response.data.decode('utf-8'))
+
+    def test_csrf_acceptance(self):
+        response = self.app.get("/admin/queryview/")
+        csrf = self.get_csrf(response)
+        response = self.app.post("/admin/queryview/", data=dict(csrf_token=csrf))
+        self.assertEqual(200, response.status_code)
+
+    def tearDown(self):
+        configuration.conf.set("webserver", "expose_config", "False")
+        self.dag_bash.clear(start_date=DEFAULT_DATE, end_date=datetime.now())
 
 class WebUiTests(unittest.TestCase):
     def setUp(self):
@@ -1415,6 +1453,7 @@ def setUp(self):
         configuration.conf.set("webserver", "expose_config", "True")
         app = application.create_app()
         app.config['TESTING'] = True
+        app.config['WTF_CSRF_METHODS'] = []
         self.app = app.test_client()
 
         self.dagbag = models.DagBag(include_examples=True)
@@ -1445,10 +1484,10 @@ def test_index(self):
     def test_query(self):
         response = self.app.get('/admin/queryview/')
         self.assertIn("Ad Hoc Query", response.data.decode('utf-8'))
-        response = self.app.get(
-            "/admin/queryview/?"
-            "conn_id=airflow_db&"
-            "sql=SELECT+COUNT%281%29+as+TEST+FROM+task_instance")
+        response = self.app.post(
+            "/admin/queryview/", data=dict(
+            conn_id="airflow_db",
+            sql="SELECT+COUNT%281%29+as+TEST+FROM+task_instance"))
         self.assertIn("TEST", response.data.decode('utf-8'))
 
     def test_health(self):
@@ -1563,9 +1602,10 @@ def test_dag_views(self):
         response = self.app.get(
             "/admin/airflow/refresh?dag_id=example_bash_operator")
         response = self.app.get("/admin/airflow/refresh_all")
-        response = self.app.get(
+        response = self.app.post(
             "/admin/airflow/paused?"
             "dag_id=example_python_operator&is_paused=false")
+        self.assertIn("OK", response.data.decode('utf-8'))
         response = self.app.get("/admin/xcom", follow_redirects=True)
         self.assertIn("Xcoms", response.data.decode('utf-8'))