codekingpro/portable-devtools

1##############################################################################2#3# pgAdmin 4 - PostgreSQL Tools4#5# Copyright (C) 2013 - 2024, The pgAdmin Development Team6# This software is released under the PostgreSQL Licence7#8##############################################################################9"""Multi-factor Authentication (MFA) views"""10 11import base6412from typing import Union13 14from flask import Response, render_template, request, flash, \15    current_app, url_for, redirect, session16from flask_babel import gettext as _17from flask_login import current_user, login_required18from flask_login.utils import login_url19 20from pgadmin.utils.csrf import pgCSRFProtect21from pgadmin.utils.ajax import bad_request22from .utils import user_supported_mfa_methods, mfa_user_registered, \23    mfa_suppored_methods, ValidationException, mfa_delete, is_mfa_enabled, \24    is_mfa_session_authenticated25from pgadmin.utils.constants import MessageType26 27 28_INDEX_URL = "browser.index"29_NO_CACHE_HEADERS = dict({30    "Cache-Control": "no-cache, no-store, must-revalidate, public, max-age=0",31    "Pragma": "no-cache",32    "Expires": "0",33})34 35 36def __handle_mfa_validation_request(37    mfa_method: str, user_mfa_auths: dict, form_data: dict38) -> None:39    """40    An internal utlity function to execute mfa.validate(...) method in case, it41    matched the following conditions:42    1. Method specified is a valid and in the supported methods list.43    2. User has registered for this auth method.44 45    Otherwise, raise an exception with appropriate error message.46 47    Args:48        mfa_method (str)     : Name of the authentication method49        user_mfa_auths (dict): List of the user supported authentication method50        form_data (dict)     : Form data in the request51 52    Raises:53        ValidationException: Raise the exception when user is not registered54                             for the given method, or not a valid MFA method.55    """56 57    if mfa_method is None:58        raise ValidationException(_("No authentication method provided."))59 60    mfa_auth = user_mfa_auths.get(mfa_method, None)61 62    if mfa_auth is None:63        raise ValidationException(_(64            "No user supported authentication method provided"65        ))66 67    mfa_auth.validate(**form_data)68 69 70@pgCSRFProtect.exempt71@login_required72def validate_view() -> Response:73    """74    An end-point to render the authentication view.75 76    It supports two HTTP methods:77    1. GET : Generate the view listing all the supported auth methods.78    2. POST: Validate the code/OTP, or whatever data the selected auth method79             supports.80 81    Returns:82        Response: Redirect to 'next' url in case authentication validate,83                  otherwise - a view with listing down all the supported auth84                  methods, and it's supporting views.85    """86 87    # Load at runtime to avoid circular dependency88    from pgadmin.authenticate import get_logout_url89 90    next_url = request.args.get("next", None)91 92    if next_url is None or next_url == url_for('mfa.register') or \93            next_url == url_for('mfa.validate'):94        next_url = url_for(_INDEX_URL)95 96    if session.get('mfa_authenticated', False) is True:97        return redirect(next_url)98 99    return_code = 200100    mfa_method = None101    user_mfa_auths = user_supported_mfa_methods()102 103    if request.method == 'POST':104        try:105            form_data = {key: request.form[key] for key in request.form}106            next_url = form_data.pop('next', url_for(_INDEX_URL))107            mfa_method = form_data.pop('mfa_method', None)108 109            __handle_mfa_validation_request(110                mfa_method, user_mfa_auths, form_data111            )112 113            session['mfa_authenticated'] = True114 115            return redirect(next_url)116 117        except ValidationException as ve:118            current_app.logger.warning((119                "MFA validation failed for the user '{}' with an error: "120                "{}"121            ).format(current_user.username, str(ve)))122            flash(str(ve), MessageType.ERROR)123            return_code = 401124        except Exception as ex:125            current_app.logger.exception(ex)126            flash(str(ex), MessageType.ERROR)127            return_code = 500128 129    mfa_views = {130        key: user_mfa_auths[key].validation_view_dict(mfa_method)131        for key in user_mfa_auths132    }133 134    if mfa_method is None and len(mfa_views) > 0:135        list(mfa_views.items())[0][1]['selected'] = True136 137    send_email_url = None138    if 'email' in mfa_views:139        send_email_url = url_for("mfa.send_email_code")140 141    return Response(render_template(142        "mfa/validate.html", _=_, views=mfa_views, base64=base64,143        logout_url=get_logout_url(),144        send_email_url=send_email_url145    ), return_code, headers=_NO_CACHE_HEADERS, mimetype="text/html")146 147 148def _mfa_registration_view(149    supported_mfa: dict, form_data: dict150) -> Union[str, None]:151    """152    An internal utility function to generate the registration view, or153    unregister for the given MFA object (passed as a dict).154 155    It will call 'registration_view' function, specific for the MFA method,156    only if User has clicked on 'Setup' button on the registration page, and157    current user is not already registered for the Auth method.158 159    If the user has not clicked on the 'Setup' button, we assume that he has160    clicked on the 'Delete' button for a specific auth method.161 162    Args:163        supported_mfa (dict): [description]164        form_data (dict): [description]165 166    Returns:167        Union[str, None]: When registration for the Auth method is completed,168                          it could return None, otherwise view for the169                          registration view.170    """171    mfa = supported_mfa['mfa']172 173    if form_data[mfa.name] == 'SETUP':174        if supported_mfa['registered'] is True:175            flash(_("'{}' is already registerd'").format(mfa.label),176                  MessageType.SUCCESS)177            return None178 179        return mfa.registration_view(form_data)180 181    if mfa_delete(mfa.name) is True:182        flash(_(183            "'{}' unregistered from the authentication list."184        ).format(mfa.label), MessageType.SUCCESS)185 186        return None187 188    flash(_(189        "'{}' is not found in the authentication list."190    ).format(mfa.label), MessageType.WARNING)191 192    return None193 194 195def _registration_view_or_deregister(196    _auth_list: dict197) -> Union[str, bool, None]:198    """199    An internal utility function to parse the request, and act upon it:200    1. Find the auth method in the request, and call the201       '_mfa_registration_view' internal utility function for the same, and202       return the result of it.203 204    It could return a registration view as a string, or None (on205    deregistering).206 207    Args:208        _auth_list (dict): List of all supported methods with a flag for the209                           current user registration.210 211    Returns:212        Union[str, bool, None]: When no valid request found, it will return213                                False, otherwise the response of the214                                '_mfa_registration_view(...)' method call.215    """216 217    for key in _auth_list:218        if key in request.form:219            return _mfa_registration_view(220                _auth_list[key], request.form221            )222 223    return False224 225 226def __handle_registration_view_for_post_method(227    _next_url: str, _mfa_auths: dict228) -> (Union[str, None], Union[Response, None], Union[dict, None]):229    """230    An internal utility function to handle the POST method for the registration231    view. It will pass on the request data to the appropriate Auth method, and232    may generate further registration view. When registration is completed, it233    will redirect to the 'next_url' in case the registration page is not opened234    from the internal dialog through menu, which can be identified by the235    'next_url' value is equal to 'internal'.236 237    Args:238        _next_url (str)  : Redirect to which url, when clicked on the239                           'continue' button on the registration page.240        _mfa_auths (dict): A dict object returned by the method -241                           'mfa_suppored_methods'.242 243    Returns:244        (Union[str, None], Union[Response, None], Union[dict, None]):245        Possibilities:246        1. Returns (None, redirect response to 'next' url, None) in case there247           is not valid 'auth' method found in the request.248        2. Returns (None, Registration view as Response, None) in case when249           valid method found, and it has returned a view to render.250        3. Otherwise - Returns the set as251           (updated 'next' url, None, updated Auth method list)252    """253 254    next_url = request.form.get("next", None)255 256    if next_url is None or next_url == url_for('mfa.validate'):257        next_url = url_for(_INDEX_URL)258 259    if request.form.get('cancel', None) is None:260        view = _registration_view_or_deregister(_mfa_auths)261 262        if view is False:263            if next_url != 'internal':264                return None, redirect(next_url), None265            flash(_("Please close the dialog."), MessageType.INFO)266 267        if view is not None:268            return None, Response(269                render_template(270                    "mfa/register.html", _=_,271                    mfa_list=list(), mfa_view=view,272                    next_url=next_url,273                    error_message=None274                ), 200,275                headers=_NO_CACHE_HEADERS276            ), None277 278        # Regenerate the supported MFA list after279        # registration/deregistration.280        _mfa_auths = mfa_suppored_methods()281 282    return next_url, None, _mfa_auths283 284 285@pgCSRFProtect.exempt286@login_required287def registration_view() -> Response:288    """289    A url end-point to register/deregister an authentication method.290 291    It supports two HTTP methods:292    * GET : Generate a view listing all the suppoted list with 'Setup',293            or 'Delete' buttons. If user has registered for the auth method, it294            will render a 'Delete' button next to it, and 'Setup' button295            otherwise.296    * POST: This handles multiple scenarios on the registration page:297            1. Clicked on the 'Delete' button, it will deregister the user for298               the specific auth method, and render the view same as for the299               'GET' method.300            2. Clicked on the 'Setup' button, it will render the registration301               view for the authentication method.302            3. Clicked 'Continue' button, redirect it to the url specified by303               'next' url.304            4. Clicking on 'Cancel' button on the Auth method specific view305               will render the view by 'GET' HTTP method.306            5. A registration method can run like a wizard, and generate307               different views based on the request data.308 309    Returns:310        Response: A response object with list of auth methods, a registration311                  view, or redirect to 'next' url312    """313    mfa_auths = mfa_suppored_methods()314    mfa_list = list()315 316    next_url = request.args.get("next", None)317 318    if request.method == 'POST':319        next_url, response, mfa_auths = \320            __handle_registration_view_for_post_method(next_url, mfa_auths)321 322        if response is not None:323            return response324 325    if next_url is None:326        next_url = url_for(_INDEX_URL)327 328    error_message = None329    found_one_mfa = False330 331    for key in mfa_auths:332        mfa = mfa_auths[key]['mfa']333        mfa = mfa.to_dict()334        mfa["registered"] = mfa_auths[key]["registered"]335        mfa_list.append(mfa)336        found_one_mfa = found_one_mfa or mfa["registered"]337 338    if request.method == 'GET':339        if is_mfa_enabled() is False:340            error_message = _(341                "Can't access this page, when multi factor authentication is "342                "disabled."343            )344        elif is_mfa_session_authenticated() is False and \345                found_one_mfa is True:346            flash(_("Complete the authentication process first"),347                  MessageType.ERROR)348            return redirect(login_url("mfa.validate", next_url=next_url))349 350    return Response(render_template(351        "mfa/register.html", _=_,352        mfa_list=mfa_list, mfa_view=None, next_url=next_url,353        error_message=error_message354    ), 200 if error_message is None else 401, headers=_NO_CACHE_HEADERS)355