When returning a Response from the oauth_error signal, use it

CHANGELOG.rst

@@ -5,6 +5,7 @@ Changelog
 -------------
 * Set ``auto_refresh_url`` automatically in ``make_azure_blueprint`` when the ``offline_access``
   scope is included, thereby enabling automatic token refresh
+* Allow returning a custom response from a ``oauth_error`` signal handler.
 
 `7.0.1`_ (2024-01-05)
 ---------------------

docs/signals.rst

@@ -80,5 +80,15 @@ The following signals exist in Flask-Dance:
     anyway, so it is your responsibility to hook into this signal and inform
     the user that there was an error.
 
+    You can also return a :class:`~flask.Response` instance from an event
+    subscriber. If you do, that response will be returned to the user instead
+    of the normal redirect. For example::
+
+        from flask import redirect, url_for
+
+        @oauth_error.connect
+        def handle_error(blueprint, error, error_description=None, error_uri=None):
+            return redirect(url_for("custom_error_page"))
+
 .. _flash a message: http://flask.pocoo.org/docs/latest/patterns/flashing/
 .. _blinker: http://pythonhosted.org/blinker/

flask_dance/consumer/oauth2.py

@@ -264,9 +264,12 @@ def authorized(self):
                 error_desc,
                 error_uri,
             )
-            oauth_error.send(
+            results = oauth_error.send(
                 self, error=error, error_description=error_desc, error_uri=error_uri
             )
+            for _, ret in results:
+                if isinstance(ret, (Response, current_app.response_class)):
+                    return ret
             return redirect(next_url)
 
         state_key = f"{self.name}_oauth_state"

tests/consumer/test_oauth2.py

@@ -768,6 +768,42 @@ def callback(*args, **kwargs):
     assert resp.status_code == 302
 
 
+@requires_blinker
+def test_signal_oauth_error_response(request):
+    app, bp = make_app()
+
+    calls = []
+
+    def callback(*args, **kwargs):
+        calls.append((args, kwargs))
+        return flask.redirect("/url")
+
+    oauth_error.connect(callback)
+    request.addfinalizer(lambda: oauth_error.disconnect(callback))
+
+    with app.test_client() as client:
+        resp = client.get(
+            "/login/test-service/authorized?"
+            "error=unauthorized_client&"
+            "error_description=Invalid+redirect+URI&"
+            "error_uri=https%3a%2f%2fexample.com%2fdocs%2fhelp",
+            base_url="https://a.b.c",
+        )
+
+        assert resp.status_code == 302
+        assert resp.headers["Location"] in ("/url", "http://localhost/url")
+
+    assert len(calls) == 1
+    assert calls[0] == (
+        (bp,),
+        {
+            "error": "unauthorized_client",
+            "error_description": "Invalid redirect URI",
+            "error_uri": "https://example.com/docs/help",
+        },
+    )
+
+
 class CustomOAuth2Session(OAuth2Session):
     my_attr = "foobar"