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 10"""A blueprint module implementing the Webserver authentication."""11 12import secrets13import string14import config15from flask import request, current_app, session, Response, render_template, \16    url_for17from flask_babel import gettext18from flask_security import login_user19from .internal import BaseAuthentication20from pgadmin.model import User21from pgadmin.tools.user_management import create_user22from pgadmin.utils.constants import WEBSERVER23from pgadmin.utils import PgAdminModule24from pgadmin.utils.csrf import pgCSRFProtect25from flask_security.utils import logout_user26 27 28class WebserverModule(PgAdminModule):29    def register(self, app, options):30        # Do not look for the sub_modules,31        # instead call blueprint.register(...) directly32        super().register(app, options)33 34    def get_exposed_url_endpoints(self):35        return ['webserver.login',36                'webserver.logout']37 38 39def init_app(app):40    MODULE_NAME = 'webserver'41 42    blueprint = WebserverModule(MODULE_NAME, __name__, static_url_path='')43 44    @blueprint.route("/login",45                     endpoint="login", methods=["GET"])46    @pgCSRFProtect.exempt47    def webserver_login():48        logout_user()49        return Response(render_template("browser/kerberos_login.html",50                                        login_url=url_for('security.login'),51                                        ))52 53    @blueprint.route("/logout",54                     endpoint="logout", methods=["GET"])55    @pgCSRFProtect.exempt56    def webserver_logout():57        logout_user()58        return Response(render_template("browser/kerberos_logout.html",59                                        login_url=url_for('security.login'),60                                        ))61 62    app.register_blueprint(blueprint)63 64 65class WebserverAuthentication(BaseAuthentication):66    LOGIN_VIEW = 'webserver.login'67    LOGOUT_VIEW = 'webserver.logout'68 69    def get_source_name(self):70        return WEBSERVER71 72    def get_friendly_name(self):73        return gettext("webserver")74 75    def validate(self, form):76        return True, None77 78    def get_user(self):79        username = request.environ.get(config.WEBSERVER_REMOTE_USER)80        if not username:81            # One more try to get the Remote User from the hearders82            username = request.headers.get(config.WEBSERVER_REMOTE_USER)83        return username84 85    def authenticate(self, form):86        username = self.get_user()87 88        if not username:89            return False, gettext(90                "Webserver authenticate failed.")91 92        session['pass_enc_key'] = ''.join(93            (secrets.choice(string.ascii_lowercase) for _ in range(10)))94        useremail = request.environ.get('mail')95        if not useremail:96            useremail = ''97        return self.__auto_create_user(username, '')98 99    def login(self, form):100        username = self.get_user()101        if username:102            user = User.query.filter_by(username=username).first()103            status = login_user(user)104            if not status:105                current_app.logger.exception(self.messages('LOGIN_FAILED'))106                return False, self.messages('LOGIN_FAILED')107            current_app.logger.info(108                "Webserver user {0} logged in.".format(username))109            return True, None110        return False, self.messages('LOGIN_FAILED')111 112    def __auto_create_user(self, username, useremail):113        """Add the webserver user to the internal SQLite database."""114        if config.WEBSERVER_AUTO_CREATE_USER:115            user = User.query.filter_by(username=username).first()116            if not user:117                create_msg = ("Creating user {0} with email {1} "118                              "from auth source Webserver.")119                current_app.logger.info(create_msg.format(username,120                                                          useremail))121                return create_user({122                    'username': username,123                    'email': useremail,124                    'role': 2,125                    'active': True,126                    'auth_source': WEBSERVER127                })128        return True, None129