fix: slight hardening

docsgpt/api/connector/routes.py

@@ -104,14 +104,17 @@ def _render_callback_page(
     """Popup page that reports an OAuth result to the opener on allowed origins only."""
     status = status if status in ("success", "error", "cancelled") else "error"
     # The script only carries server-side values: the provider key comes from the
-    # supported-connector list rather than the request, and nothing posts without a token.
+    # supported-connector list rather than the request, and no request text is posted.
     provider_key = next(
         (key for key in ConnectorCreator.get_supported_connectors() if key == provider_raw.lower()), None,
     )
-    payload = (
-        {"type": f"{provider_key}_auth_success", "session_token": session_token, "user_email": user_email}
-        if status == "success" and session_token and provider_key else None
-    )
+    payload = None
+    if provider_key and status == "success" and session_token:
+        payload = {"type": f"{provider_key}_auth_success", "session_token": session_token, "user_email": user_email}
+    elif provider_key and status == "error":
+        # The frontend shows its own localized failure message; cancellations are
+        # reported when the popup closes.
+        payload = {"type": f"{provider_key}_auth_error"}
     target_origins = connector_allowed_origins(request.host_url) if payload else []
     provider = html.escape(provider_raw.replace("_", " ").title())
     connected_as = (

tests/api/test_connector_oauth_security.py

@@ -145,6 +145,31 @@ class TestCallbackStatusPage:
         script = body[body.index("<script>"):body.index("</script>")]
         assert "zz-injected-provider" not in script
 
+    def test_error_posts_fixed_auth_error_to_allowed_origins(self, app):
+        from docsgpt.api.connector.routes import ConnectorCallbackStatus
+        from docsgpt.core.settings import settings
+
+        with patch.object(settings, "CONNECTOR_ALLOWED_ORIGINS", "https://app.example.com"), app.test_request_context(
+            "/api/connectors/callback-status?status=error&provider=google_drive&message=zz-request-message"
+        ):
+            r = ConnectorCallbackStatus().get()
+        body = r.get_data(as_text=True)
+        script = body[body.index("<script>"):body.index("</script>")]
+        assert '{"type": "google_drive_auth_error"}' in script
+        assert '"https://app.example.com"' in script
+        assert "zz-request-message" not in script
+
+    def test_cancelled_posts_nothing(self, app):
+        from docsgpt.api.connector.routes import ConnectorCallbackStatus
+
+        with app.test_request_context(
+            "/api/connectors/callback-status?status=cancelled&provider=google_drive"
+        ):
+            r = ConnectorCallbackStatus().get()
+        body = r.get_data(as_text=True)
+        assert "const payload = null;" in body
+        assert "const targetOrigins = [];" in body
+
 
 class TestCallbackDeliversTokenSafely:
     def test_success_renders_page_without_redirecting_token(self, app, pg_conn):
@@ -360,6 +385,27 @@ class TestRemoteUploadSessionOwnership:
         apply_mock.assert_not_called()
 
 
+class TestValidateSessionOwnership:
+    def test_rejects_foreign_session_token(self, app, pg_conn):
+        from docsgpt.api.connector.routes import ConnectorValidateSession
+
+        _seed_session(pg_conn, "u-victim-val", "st-victim-val", token_info={"access_token": "victim-at"})
+        create_auth = MagicMock()
+        with _patch_db(pg_conn), patch(
+            "docsgpt.api.connector.routes.ConnectorCreator.create_auth", create_auth,
+        ), app.test_request_context(
+            "/api/connectors/validate-session", method="POST",
+            json={"provider": "google_drive", "session_token": "st-victim-val"},
+        ):
+            from flask import request
+            request.decoded_token = {"sub": "u-attacker-val"}
+            r = ConnectorValidateSession().post()
+
+        assert r.status_code == 401
+        assert "victim-at" not in r.get_data(as_text=True)
+        create_auth.assert_not_called()
+
+
 class TestSessionProviderBinding:
     def _files(self, app, pg_conn, user, provider, token, create_connector):
         from docsgpt.api.connector.routes import ConnectorFiles