CVE-2021-32805 (Python, published 2021-09-08T18:15Z)
Description: value: Flask-AppBuilder is an application development framework, built on top of Flask. In affected versions if using Flask-AppBuilder OAuth, an attacker can share a carefully crafted URL with a trusted domain for an application built with Flask-AppBuilder, this URL can redirect a user to a malicious site. This is an open redirect vulnerability. To resolve this issue upgrade to Flask-AppBuilder 3.2.2 or above. If upgrading is infeasible users may filter HTTP traffic containing `?next={next-site}` where the `next-site` domain is different from the application you are protecting as a workaround.
Severity: MEDIUM (CVSS v3 base score: 6.1)
Weakness: CWE-601 \u2014 URL Redirection to Untrusted Site ('Open Redirect')
A web application accepts a user-controlled input that specifies a link to an external site, and uses that link in a Redirect. This simplifies phishing attacks.
Repository: https://github.com/dpgaspar/Flask-AppBuilder
Files changed:
flask_appbuilder/security/views.py
flask_appbuilder/tests/config_oauth.py
flask_appbuilder/tests/test_mvc_oauth.py
requirements-extra.txt
setup.py
Commit message: fix: improve next URL on OAuth (#1668)

* fix: improve next URL on OAuth

* add tests and extra dependency for OAuth

* lint

* fix test

* add test for unknown provider

* lint

* Update flask_appbuilder/security/views.py

Co-authored-by: Ash Berlin-Taylor <ash_github@firemirror.com>

Co-authored-by: Ash Berlin-Taylor <ash_github@firemirror.com>

Vulnerable code:

from flask import abort, current_app, flash, g, redirect, request, session, url_for
    """
        For Future Use, API Auth, must check howto keep REST stateless
    """

    """
    @expose_api(name='auth',url='/api/auth')
    def auth(self):
        if g.user is not None and g.user.is_authenticated:
            http_return_code = 401
            response = make_response(
                jsonify(
                    {
                        'message': 'Login Failed already authenticated',
                        'severity': 'critical'
                    }
                ),
                http_return_code
            )
        username = str(request.args.get('username'))
        password = str(request.args.get('password'))
        user = self.appbuilder.sm.auth_user_ldap(username, password)
        if not user:
            http_return_code = 401
            response = make_response(
                jsonify(
                    {
                        'message': 'Login Failed',
                        'severity': 'critical'
                    }
                ),
                http_return_code
            )
        else:
            login_user(user, remember=False)
            http_return_code = 201
            response = make_response(
                jsonify(
                    {
                        'message': 'Login Success',
                         'severity': 'info'
                    }
                ),
                http_return_code
            )
        return response
    """

    def login(self, provider=None, register=None):
    def oauth_authorized(self, provider):
            try:
                next_url = state["next"][0] or self.appbuilder.get_url_for_index
            except (KeyError, IndexError):
                next_url = self.appbuilder.get_url_for_index

Authlib==0.15.2
    extras_require={"jmespath": ["jmespath>=0.9.5"]},

Fixed code:
from typing import Optional
from urllib.parse import urlparse

from flask import (
    abort,
    current_app,
    flash,
    g,
    redirect,
    request,
    Response,
    session,
    url_for,
)
    def login(
        self, provider: Optional[str] = None, register: Optional[str] = None
    ) -> Response:
    def oauth_authorized(self, provider: str) -> Response:
        if provider not in self.appbuilder.sm.oauth_remotes:
            flash(u"Provider not supported.", "warning")
            log.warning("OAuth authorized got an unknown provider %s", provider)
            return redirect(self.appbuilder.get_url_for_login)
            next_url = self.appbuilder.get_url_for_index
            # Check if there is a next url on state
            if "next" in state and len(state["next"]) > 0:
                parsed_uri = urlparse(state["next"][0])
                if parsed_uri.netloc != request.host:
                    log.warning("Got an invalid next URL: %s", parsed_uri.netloc)
                else:
                    next_url = state["next"][0]
import os

from flask_appbuilder.security.manager import AUTH_OAUTH

basedir = os.path.abspath(os.path.dirname(__file__))

SQLALCHEMY_DATABASE_URI = os.environ.get(
    "SQLALCHEMY_DATABASE_URI"
) or "sqlite:///" + os.path.join(basedir, "app.db")

SECRET_KEY = "thisismyscretkey"

AUTH_TYPE = AUTH_OAUTH

OAUTH_PROVIDERS = [
    {
        "name": "google",
        "icon": "fa-google",
        "token_key": "access_token",
        "remote_app": {
            "client_id": "CLIENT_ID",
            "client_secret": "CLIENT_SECRET",
            "api_base_url": "https://www.googleapis.com/oauth2/v2/",
            "client_kwargs": {"scope": "email profile"},
            "request_token_url": None,
            "access_token_url": "https://accounts.google.com/o/oauth2/token",
            "authorize_url": "https://accounts.google.com/o/oauth2/auth",
        },
    }
]

# Will allow user self registration
AUTH_USER_REGISTRATION = True

# The default user self registration role for all users
AUTH_USER_REGISTRATION_ROLE = "Admin"
from flask_appbuilder import SQLA
from flask_appbuilder.tests.base import FABTestCase
import jwt


class UserInfoReponseMock:
    def json(self):
        return {
            "id": "1",
            "given_name": "first-name",
            "family_name": "last-name",
            "email": "user1@fab.org",
        }


class OAuthRemoteMock:
    def authorize_access_token(self):
        return {"access_token": "some-key"}

    def get(self, item):
        if item == "userinfo":
            return UserInfoReponseMock()


class APICSRFTestCase(FABTestCase):
    def setUp(self):
        from flask import Flask
        from flask_wtf import CSRFProtect
        from flask_appbuilder import AppBuilder

        self.app = Flask(__name__)
        self.app.config.from_object("flask_appbuilder.tests.config_oauth")
        self.app.config["WTF_CSRF_ENABLED"] = True

        self.csrf = CSRFProtect(self.app)
        self.db = SQLA(self.app)
        self.appbuilder = AppBuilder(self.app, self.db.session)

    def test_oauth_login(self):
        """
        OAuth: Test login
        """
        client = self.app.test_client()

        self.appbuilder.sm.oauth_remotes = {"google": OAuthRemoteMock()}

        raw_state = {}
        state = jwt.encode(raw_state, self.app.config["SECRET_KEY"], algorithm="HS256")

        response = client.get(f"/oauth-authorized/google?state={state.decode('utf-8')}")
        self.assertEqual(response.location, "http://localhost/")

    def test_oauth_login_unknown_provider(self):
        """
        OAuth: Test login with unknown provider
        """
        client = self.app.test_client()

        self.appbuilder.sm.oauth_remotes = {"google": OAuthRemoteMock()}

        raw_state = {}
        state = jwt.encode(raw_state, self.app.config["SECRET_KEY"], algorithm="HS256")

        response = client.get(
            f"/oauth-authorized/unknown_provider?state={state.decode('utf-8')}"
        )
        self.assertEqual(response.location, "http://localhost/login/")

    def test_oauth_login_next(self):
        """
        OAuth: Test login next
        """
        client = self.app.test_client()

        self.appbuilder.sm.oauth_remotes = {"google": OAuthRemoteMock()}

        raw_state = {"next": ["http://localhost/users/list/"]}
        state = jwt.encode(raw_state, self.app.config["SECRET_KEY"], algorithm="HS256")

        response = client.get(f"/oauth-authorized/google?state={state.decode('utf-8')}")
        self.assertEqual(response.location, "http://localhost/users/list/")

    def test_oauth_login_next_check(self):
        """
        OAuth: Test login next check
        """
        client = self.app.test_client()

        self.appbuilder.sm.oauth_remotes = {"google": OAuthRemoteMock()}

        raw_state = {"next": ["http://www.google.com"]}
        state = jwt.encode(raw_state, self.app.config["SECRET_KEY"], algorithm="HS256")

        response = client.get(f"/oauth-authorized/google?state={state.decode('utf-8')}")
        self.assertEqual(response.location, "http://localhost/")
Authlib==0.15.4
    extras_require={
        "jmespath": ["jmespath>=0.9.5"],
        "oauth": ["Authlib>=0.14, <1.0.0"],
    },