feat(device-flow): redirect SSO-complete failures to a dedicated device error view (#38185)

api/tests/unit_tests/controllers/openapi/test_device_sso.py

@@ -1,6 +1,7 @@
 """SSO-branch device-flow endpoints under /openapi/v1/oauth/device/."""
 
 import builtins
+from unittest.mock import MagicMock, patch
 
 import pytest
 from flask import Flask
@@ -77,3 +78,97 @@ def test_sso_complete_idp_callback_url_uses_canonical_path():
     from controllers.openapi import oauth_device_sso
 
     assert oauth_device_sso._SSO_COMPLETE_PATH == "/openapi/v1/oauth/device/sso-complete"
+
+
+# ---------------------------------------------------------------------------
+# _device_error_redirect helper
+# ---------------------------------------------------------------------------
+
+
+def test_device_error_redirect_builds_relative_location():
+    from controllers.openapi import oauth_device_sso
+
+    app = Flask(__name__)
+    with app.test_request_context():
+        resp = oauth_device_sso._device_error_redirect("sso_failed", "ABCD-1234")
+    assert resp.status_code == 302
+    loc = resp.headers["Location"]
+    assert loc.startswith("/device?")
+    assert "sso_error=sso_failed" in loc
+    assert "user_code=ABCD-1234" in loc
+
+
+def test_device_error_redirect_clamps_unknown_code():
+    from controllers.openapi import oauth_device_sso
+
+    app = Flask(__name__)
+    with app.test_request_context():
+        resp = oauth_device_sso._device_error_redirect("totally-bogus")
+    assert "sso_error=sso_failed" in resp.headers["Location"]
+
+
+def test_device_error_redirect_keeps_email_special_case():
+    from controllers.openapi import oauth_device_sso
+
+    app = Flask(__name__)
+    with app.test_request_context():
+        resp = oauth_device_sso._device_error_redirect("email_belongs_to_dify_account", "ABCD-1234")
+    assert "sso_error=email_belongs_to_dify_account" in resp.headers["Location"]
+
+
+def test_device_error_redirect_omits_empty_user_code():
+    from controllers.openapi import oauth_device_sso
+
+    app = Flask(__name__)
+    with app.test_request_context():
+        resp = oauth_device_sso._device_error_redirect("sso_failed")
+    assert "user_code=" not in resp.headers["Location"]
+
+
+def test_device_error_redirect_drops_malformed_user_code():
+    from controllers.openapi import oauth_device_sso
+
+    app = Flask(__name__)
+    with app.test_request_context():
+        resp = oauth_device_sso._device_error_redirect("sso_failed", "https://evil.example/")
+    loc = resp.headers["Location"]
+    assert loc.startswith("/device?")
+    assert "user_code=" not in loc
+    assert "evil" not in loc
+
+
+# ---------------------------------------------------------------------------
+# sso_complete redirect behaviour
+# ---------------------------------------------------------------------------
+
+
+def _ee_features():
+    from services.feature_service import LicenseStatus
+
+    m = MagicMock()
+    m.license.status = LicenseStatus.ACTIVE
+    return m
+
+
+@patch("libs.device_flow_security.FeatureService.get_system_features")
+def test_sso_complete_relays_inbound_sso_error(ee_feat, openapi_app):
+    ee_feat.return_value = _ee_features()
+    client = openapi_app.test_client()
+    resp = client.get(
+        "/openapi/v1/oauth/device/sso-complete?sso_error=sso_failed&user_code=ABCD-1234",
+        follow_redirects=False,
+    )
+    assert resp.status_code == 302
+    loc = resp.headers["Location"]
+    assert "/device?" in loc
+    assert "sso_error=sso_failed" in loc
+    assert "user_code=ABCD-1234" in loc
+
+
+@patch("libs.device_flow_security.FeatureService.get_system_features")
+def test_sso_complete_missing_assertion_redirects_generic(ee_feat, openapi_app):
+    ee_feat.return_value = _ee_features()
+    client = openapi_app.test_client()
+    resp = client.get("/openapi/v1/oauth/device/sso-complete", follow_redirects=False)
+    assert resp.status_code == 302
+    assert "sso_error=sso_failed" in resp.headers["Location"]

api/tests/unit_tests/controllers/openapi/test_oauth_sso_claims.py

@@ -34,8 +34,9 @@ def test_sso_complete_rejects_assertion_missing_email(ee_feat, jws_mod, app: Fla
     jws_mod.VerifyError = Exception
 
     client = app.test_client()
-    resp = client.get("/openapi/v1/oauth/device/sso-complete?sso_assertion=blob")
-    assert resp.status_code == 400, resp.data
+    resp = client.get("/openapi/v1/oauth/device/sso-complete?sso_assertion=blob", follow_redirects=False)
+    assert resp.status_code == 302, resp.data
+    assert "sso_error=sso_failed" in resp.headers["Location"]
 
 
 @patch("controllers.openapi.oauth_device_sso.jws")
@@ -48,8 +49,9 @@ def test_sso_complete_rejects_assertion_empty_issuer(ee_feat, jws_mod, app: Flas
     jws_mod.VerifyError = Exception
 
     client = app.test_client()
-    resp = client.get("/openapi/v1/oauth/device/sso-complete?sso_assertion=blob")
-    assert resp.status_code == 400
+    resp = client.get("/openapi/v1/oauth/device/sso-complete?sso_assertion=blob", follow_redirects=False)
+    assert resp.status_code == 302
+    assert "sso_error=sso_failed" in resp.headers["Location"]
 
 
 def test_verify_approval_grant_raises_on_missing_field():