Repository: streamlit/streamlit
File: lib/streamlit/web/server/oidc_mixin.py (Python)

Code before:
        :param kwargs: Extra parameters to include.
        :return: A HTTP redirect response.
        """
        auth_context = self.create_authorization_url(redirect_uri, **kwargs)
        self._save_authorize_data(redirect_uri=redirect_uri, **auth_context)
        request_handler.redirect(auth_context["url"], status=302)

    def authorize_access_token(
        self, request_handler: tornado.web.RequestHandler, **kwargs: Any
    ) -> dict[str, Any]:
        """
        :param request_handler: HTTP request instance from Tornado.
        :return: A token dict.
        """
        error = request_handler.get_argument("error", None)
        if error:
            description = request_handler.get_argument("error_description", None)
            raise OAuthError(error=error, description=description)

        params = {
            "code": request_handler.get_argument("code"),
            "state": request_handler.get_argument("state"),
        }

        # Tornado doesn't have sessions like Flask/Django, so we use an empty dict.
        # Authlib 1.6.6+ always writes state to session even when cache is available.
        session: dict[str, Any] = {}

        claims_options = kwargs.pop("claims_options", None)
        state_data = self.framework.get_state_data(session, params.get("state"))
        self.framework.clear_state_data(session, params.get("state"))

        if not state_data:
            raise OAuthError(
                error="invalid_state",
                description="OAuth state not found or expired. Please try logging in again.",
            )

        params = self._format_state_params(state_data, params)  # type: ignore[attr-defined]
        token = self.fetch_access_token(**params, **kwargs)

        if "id_token" in token and "nonce" in state_data:
            userinfo = self.parse_id_token(
                token, nonce=state_data["nonce"], claims_options=claims_options
            )
            token = {**token, "userinfo": userinfo}
        return cast("dict[str, Any]", token)

    def _save_authorize_data(self, **kwargs: Any) -> None:
        """Authlib underlying uses the concept of "session" to store state data.
        In Tornado, we don't have a session, so we use an empty dict as a placeholder.

Review comment (security):
```suggestion
        # Authlib 1.6.6+ always writes state to session even when cache is available. We also override state access to use the cache instead of session in `TornadoIntegration`.
```

Code after:
        :param kwargs: Extra parameters to include.
        :return: A HTTP redirect response.
        """
        auth_context = self.create_authorization_url(redirect_uri, **kwargs)
        self._save_authorize_data(redirect_uri=redirect_uri, **auth_context)
        request_handler.redirect(auth_context["url"], status=302)

    def authorize_access_token(
        self, request_handler: tornado.web.RequestHandler, **kwargs: Any
    ) -> dict[str, Any]:
        """
        :param request_handler: HTTP request instance from Tornado.
        :return: A token dict.
        """
        error = request_handler.get_argument("error", None)
        if error:
            description = request_handler.get_argument("error_description", None)
            raise OAuthError(error=error, description=description)

        params = {
            "code": request_handler.get_argument("code"),
            "state": request_handler.get_argument("state"),
        }

        # Authlib 1.6.6+ always writes state to session even when cache is available.
        # We also override state access to use the cache instead of session in `TornadoIntegration`.
        session: dict[str, Any] = {}

        claims_options = kwargs.pop("claims_options", None)
        state_data = self.framework.get_state_data(session, params.get("state"))
        self.framework.clear_state_data(session, params.get("state"))

        if not state_data:
            raise OAuthError(
                error="invalid_state",
                description="OAuth state not found or expired. Please try logging in again.",
            )

        params = self._format_state_params(state_data, params)  # type: ignore[attr-defined]
        token = self.fetch_access_token(**params, **kwargs)

        if "id_token" in token and "nonce" in state_data:
            userinfo = self.parse_id_token(
                token, nonce=state_data["nonce"], claims_options=claims_options
            )
            token = {**token, "userinfo": userinfo}
        return cast("dict[str, Any]", token)

    def _save_authorize_data(self, **kwargs: Any) -> None:
        """Authlib underlying uses the concept of "session" to store state data.
        In Tornado, we don't have a session, so we use an empty dict as a placeholder.