[#7028] Disable CSRF protection when authenticating via header

ckan/config/middleware/flask_app.py

@@ -327,7 +327,10 @@ def load_user_from_request(request):  # type: ignore
         This callback function is called whenever a user could not be
         authenticated via the session cookie, so we fall back to the API token.
         """
+        g.login_via_request = True
+
         user = _get_user_for_apitoken()
+
         return user
 
     # Update the main CKAN config object with the Flask specific keys
@@ -410,6 +413,11 @@ def ckan_before_request() -> Optional[Response]:
     # Sets g.user and g.userobj for extensions
     response = identify_user()
 
+    # Disable CSRF protection if user was logged in via the Authorization
+    # header from the request
+    if g.get("login_via_request"):
+        csrf.exempt(f"ckan.views.{request.endpoint}")
+
     # Set the csrf_field_name so we can use it in our templates
     g.csrf_field_name = config.get_value("WTF_CSRF_FIELD_NAME")
 

ckan/tests/controllers/test_api.py

@@ -380,3 +380,38 @@ def test_cookie_based_auth_disabled(app):
     res = app.get(url, environ_overrides=env)
 
     assert res.status_code == 200
+
+
+@pytest.mark.usefixtures("clean_db")
+def test_header_based_auth_default(app):
+
+    sysadmin = factories.SysadminWithToken()
+    org = factories.Organization()
+    dataset = factories.Dataset(private=True, owner_org=org["id"])
+
+    url = url_for("api.action", ver=3, logic_function="package_show", id=dataset["id"])
+
+    env = {"Authorization": sysadmin["token"]}
+
+    res = app.get(url, environ_overrides=env)
+
+    assert res.status_code == 200
+
+@pytest.mark.usefixtures("clean_db")
+def test_header_based_auth_default_post(app):
+
+    sysadmin = factories.SysadminWithToken()
+    org = factories.Organization()
+    dataset = factories.Dataset(private=True, owner_org=org["id"])
+
+    url = url_for("api.action", ver=3, logic_function="package_patch")
+
+    env = {"Authorization": sysadmin["token"]}
+
+    data = {
+        "id": dataset["id"],
+        "notes": "updated",
+    }
+    res = app.post(url, environ_overrides=env, data=data)
+
+    assert res.status_code == 200