__init__.py

##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2026, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################

import json
from collections import OrderedDict
import pgadmin.browser.server_groups as sg
from flask import render_template, request, make_response, jsonify, \
    current_app, url_for, session
from flask_babel import gettext
from flask_security import current_user, permissions_required
from pgadmin.user_login_check import pga_login_required
from psycopg.conninfo import make_conninfo, conninfo_to_dict

from pgadmin.browser.server_groups.servers.types import ServerType
from pgadmin.browser.utils import PGChildNodeView
from pgadmin.utils.ajax import make_json_response, bad_request, forbidden, \
    make_response as ajax_response, internal_server_error, unauthorized, gone
from pgadmin.utils.crypto import encrypt, decrypt
from pgadmin.utils.menu import MenuItem
from pgadmin.tools.sqleditor.utils.query_history import QueryHistory
from pgadmin.tools.user_management.PgAdminPermissions import AllPermissionTypes

import config
from config import PG_DEFAULT_DRIVER
from pgadmin.model import db, Server, ServerGroup, User, SharedServer
from pgadmin.utils.driver import get_driver
from pgadmin.utils.master_password import get_crypt_key
from pgadmin.utils.exception import CryptKeyMissing, ConnectionLost
from pgadmin.tools.schema_diff.node_registry import SchemaDiffRegistry
from pgadmin.browser.server_groups.servers.utils import \
    (is_valid_ipaddress, get_replication_type, convert_connection_parameter,
     check_ssl_fields, get_db_restriction)
from pgadmin.utils.constants import UNAUTH_REQ, MIMETYPE_APP_JS, \
    SERVER_CONNECTION_CLOSED, RESTRICTION_TYPE_SQL
from sqlalchemy import or_
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import object_session
from sqlalchemy.orm.attributes import flag_modified
from pgadmin.utils.preferences import Preferences
from .... import socketio as sio
from pgadmin.utils import get_complete_file_path
from pgadmin.settings.utils import with_object_filters
from pgadmin.utils.server_access import get_server, \
    get_user_server_query, get_server_group


# File-path keys in connection_params that are per-user and must
# not be copied from the owner to a new SharedServer or leaked
# through the property merge.
SENSITIVE_CONN_KEYS = frozenset({
    'passfile', 'sslcert', 'sslkey',
    'sslrootcert', 'sslcrl', 'sslcrldir',
})


def _is_non_owner(server):
    """True if the server is shared and the current user is not
    the owner.  Centralises the check used in 15+ places."""
    return server.shared and server.user_id != current_user.id


def has_any(data, keys):
    """
    Checks any one of the keys present in the data given
    """
    if data is None and not isinstance(data, dict):
        return False

    if keys is None and not isinstance(keys, list):
        return False

    for key in keys:
        if key in data:
            return True

    return False


def recovery_state(connection, postgres_version):
    recovery_check_sql = render_template(
        "connect/sql/#{0}#/check_recovery.sql".format(postgres_version))

    status, result = connection.execute_dict(recovery_check_sql)
    if status and 'rows' in result and len(result['rows']) > 0:
        in_recovery = result['rows'][0]['inrecovery']
        wal_paused = result['rows'][0]['isreplaypaused']
    else:
        in_recovery = None
        wal_paused = None
    return status, result, in_recovery, wal_paused


def get_preferences():
    """
    Get preferences setting
    :return: whether to hide shared server or not.
    """
    hide_shared_server = None
    if config.SERVER_MODE:
        pref = Preferences.module('browser')
        hide_shared_server = pref.preference('hide_shared_server').get()

    return hide_shared_server


def server_icon_and_background(is_connected, manager, server):
    """

    Args:
        is_connected: Flag to check if server is connected
        manager: Connection manager
        server: Sever object

    Returns:
        Server Icon CSS class
    """
    server_background_color = ''
    if server and server.bgcolor:
        server_background_color = ' {0}'.format(
            server.bgcolor
        )
        # If user has set font color also
        if server.fgcolor:
            server_background_color = '{0} {1}'.format(
                server_background_color,
                server.fgcolor
            )

    if is_connected:
        return 'icon-{0}{1}'.format(
            manager.server_type, server_background_color
        )
    elif server.shared and config.SERVER_MODE:
        return 'icon-shared-server-not-connected{0}'.format(
            server_background_color
        )
    elif server.cloud_status == -1:
        return 'icon-server-cloud-deploy{0}'.format(
            server_background_color
        )
    else:
        return 'icon-server-not-connected{0}'.format(
            server_background_color
        )


class ServerModule(sg.ServerGroupPluginModule):
    _NODE_TYPE = "server"
    LABEL = gettext("Servers")

    @property
    def node_type(self):
        return self._NODE_TYPE

    @property
    def script_load(self):
        """
        Load the module script for server, when any of the server-group node is
        initialized.
        """
        return sg.ServerGroupModule.node_type

    @staticmethod
    def get_shared_server_properties(server, sharedserver):
        """
        Return shared server properties.

        Overlays per-user SharedServer values onto the owner's Server
        object so each non-owner sees their own customizations.

        The server is expunged from the SQLAlchemy session before
        mutation so that the owner's record is never dirtied.
        :param server:
        :param sharedserver:
        :return: shared server (detached)
        """
        if sharedserver is None:
            return server

        # Detach from session so in-place mutations are never
        # flushed back to the owner's Server row.
        sess = object_session(server)
        if sess is not None:
            sess.expunge(server)

        server.bgcolor = sharedserver.bgcolor
        server.fgcolor = sharedserver.fgcolor
        server.name = sharedserver.name
        server.role = sharedserver.role
        server.service = sharedserver.service
        server.use_ssh_tunnel = sharedserver.use_ssh_tunnel
        server.tunnel_host = sharedserver.tunnel_host
        server.tunnel_port = sharedserver.tunnel_port
        server.tunnel_authentication = sharedserver.tunnel_authentication
        server.tunnel_username = sharedserver.tunnel_username
        server.tunnel_password = sharedserver.tunnel_password
        server.save_password = sharedserver.save_password
        server.tunnel_identity_file = sharedserver.tunnel_identity_file
        server.tunnel_prompt_password = sharedserver.tunnel_prompt_password

        # Override per-user connection_params keys.  Use the
        # SharedServer value whenever it is present, regardless of
        # whether the owner's Server has the same key.
        s_conn = getattr(server, 'connection_params', None) \
            or {}
        ss_conn = getattr(sharedserver, 'connection_params',
                          None) or {}
        for key in SENSITIVE_CONN_KEYS:
            if key in ss_conn:
                s_conn[key] = ss_conn[key]
            elif key in s_conn:
                # Owner has this key but non-owner doesn't —
                # remove it so the owner's path doesn't leak.
                del s_conn[key]
        server.connection_params = s_conn

        server.servergroup_id = sharedserver.servergroup_id
        server.username = sharedserver.username
        server.server_owner = sharedserver.server_owner
        server.password = sharedserver.password
        server.prepare_threshold = sharedserver.prepare_threshold
        server.passexec_cmd = sharedserver.passexec_cmd
        server.passexec_expiration = sharedserver.passexec_expiration
        server.kerberos_conn = sharedserver.kerberos_conn
        server.tags = sharedserver.tags
        server.post_connection_sql = sharedserver.post_connection_sql

        return server

    def get_servers(self, all_servers, hide_shared_server, gid):
        """
        This function creates list of servers which needs to display
        in browser tree
        :param all_servers:
        :param hide_shared_server:
        :param gid:
        :return: list of servers
        """
        servers = []
        for server in all_servers:
            if server.discovery_id and \
                not server.shared and \
                config.SERVER_MODE and \
                SharedServer.query.filter_by(
                    user_id=current_user.id,
                    osid=server.id).first() is not None \
                    and not hide_shared_server:
                continue

            if _is_non_owner(server):

                shared_server = self.get_shared_server(server, gid)

                if hide_shared_server:
                    # Don't include shared server if hide shared server is
                    # set to true.
                    continue

                server = self.get_shared_server_properties(server,
                                                           shared_server)
            servers.append(server)

        return servers

    def has_tag(self, server, object_filters):
        try:
            # No tags filter, show all
            if len(object_filters['tags']) == 0:
                return True

            # No tags on server, don't show
            if server.tags is None or len(server.tags) == 0:
                return False

            # Check if any of the tag exists
            return any([t['text'] in object_filters['tags']
                        for t in server.tags])
        except Exception as _:
            return True

    @with_object_filters
    @pga_login_required
    def get_nodes(self, gid, object_filters):
        """Return a JSON document listing the server groups for the user"""

        hide_shared_server = get_preferences()
        servers = get_user_server_query().filter(
            Server.servergroup_id == gid, Server.is_adhoc == 0)

        driver = get_driver(PG_DEFAULT_DRIVER)
        servers = self.get_servers(servers, hide_shared_server, gid)

        for server in servers:
            connected = False
            manager = None
            errmsg = None
            was_connected = False
            in_recovery = None
            wal_paused = None
            server_type = 'pg'
            user_info = None

            if not self.has_tag(server, object_filters):
                continue

            try:
                manager = driver.connection_manager(server.id)
                conn = manager.connection()
                was_connected = conn.wasConnected
                connected = conn.connected()
                if connected:
                    server_type = manager.server_type
                    user_info = manager.user_info
            except CryptKeyMissing:
                # show the nodes at least even if not able to connect.
                pass
            except Exception as e:
                current_app.logger.exception(e)
                errmsg = str(e)
            yield self.generate_browser_node(
                "%d" % (server.id),
                gid,
                server.name,
                server_icon_and_background(connected, manager, server),
                True,
                self.node_type,
                connected=connected,
                server_type=server_type,
                version=manager.version,
                db=manager.db,
                user=user_info,
                in_recovery=in_recovery,
                wal_pause=wal_paused,
                host=server.host,
                port=server.port,
                is_password_saved=bool(server.save_password),
                is_tunnel_password_saved=bool(server.tunnel_password),
                was_connected=was_connected,
                errmsg=errmsg,
                user_id=server.user_id,
                username=server.username,
                shared=server.shared,
                is_kerberos_conn=bool(server.kerberos_conn),
                gss_authenticated=manager.gss_authenticated,
                cloud_status=server.cloud_status,
                description=server.comment,
                tags=server.tags
            )

    @property
    def jssnippets(self):
        return []

    @property
    def csssnippets(self):
        """
        Returns a snippet of css to include in the page
        """
        snippets = [render_template("css/servers.css")]

        for submodule in self.submodules:
            snippets.extend(submodule.csssnippets)

        for st in ServerType.types():
            snippets.extend(st.csssnippets)

        return snippets

    def register(self, app, options):
        """
        Override the default register function to automagically register
        sub-modules at once.
        """
        driver = get_driver(PG_DEFAULT_DRIVER, app)
        app.jinja_env.filters['qtLiteral'] = driver.qtLiteral
        app.jinja_env.filters['qtIdent'] = driver.qtIdent
        app.jinja_env.filters['qtTypeIdent'] = driver.qtTypeIdent
        app.jinja_env.filters['hasAny'] = has_any

        from .ppas import PPAS

        from .databases import blueprint as module
        self.submodules.append(module)

        from .pgagent import blueprint as module
        self.submodules.append(module)

        from .resource_groups import blueprint as module
        self.submodules.append(module)

        from .roles import blueprint as module
        self.submodules.append(module)

        from .tablespaces import blueprint as module
        self.submodules.append(module)

        from .directories import blueprint as module
        self.submodules.append(module)

        from .replica_nodes import blueprint as module
        self.submodules.append(module)

        from .pgd_replication_groups import blueprint as module
        self.submodules.append(module)

        super().register(app, options)

    # We do not have any preferences for server node.
    def register_preferences(self):
        """
        register_preferences
        Override it so that - it does not register the show_node preference for
        server type.
        """
        ServerType.register_preferences()

    def get_exposed_url_endpoints(self):
        return ['NODE-server.connect_id']

    @staticmethod
    def create_shared_server(data, gid):
        """
        Create shared server
        :param data:
        :param gid:
        :return: None
        """

        shared_server = None
        try:
            db.session.rollback()
            user = User.query.filter_by(id=data.user_id).first()

            # Strip owner's sensitive file paths from
            # connection_params — each user should configure
            # their own SSL/passfile paths.
            safe_conn_params = {}
            if data.connection_params:
                safe_conn_params = {
                    k: v for k, v in
                    data.connection_params.items()
                    if k not in SENSITIVE_CONN_KEYS
                }

            shared_server = SharedServer(
                osid=data.id,
                user_id=current_user.id,
                server_owner=user.username,
                servergroup_id=gid,
                name=data.name,
                host=data.host,
                port=data.port,
                maintenance_db=data.maintenance_db,
                username=data.shared_username,
                save_password=0,
                comment=None,
                role=data.role,
                bgcolor=data.bgcolor if data.bgcolor else None,
                fgcolor=data.fgcolor if data.fgcolor else None,
                service=data.service if data.service else None,
                use_ssh_tunnel=data.use_ssh_tunnel,
                tunnel_host=data.tunnel_host,
                tunnel_port=data.tunnel_port
                if data.tunnel_port else 22,
                tunnel_username=None,
                tunnel_authentication=0,
                tunnel_identity_file=None,
                tunnel_keep_alive=data.tunnel_keep_alive
                if data.tunnel_keep_alive else 0,
                tunnel_prompt_password=0,
                shared=True,
                connection_params=safe_conn_params,
                prepare_threshold=data.prepare_threshold,
                passexec_cmd=None,
                passexec_expiration=None,
                kerberos_conn=False,
                tags=None,
                post_connection_sql=None
            )
            db.session.add(shared_server)
            db.session.commit()
        except Exception as e:
            db.session.rollback()
            raise e

    @staticmethod
    def get_shared_server(server, gid):
        """
        Return the SharedServer record for the current user,
        creating one lazily if it doesn't exist.  The unique
        constraint on (osid, user_id) prevents duplicates from
        concurrent requests.
        :param server:
        :param gid:
        :return: shared_server (never None)
        :raises: Exception if SharedServer cannot be created
        """
        shared_server = SharedServer.query.filter_by(
            user_id=current_user.id,
            osid=server.id).first()

        if shared_server is None:
            try:
                ServerModule.create_shared_server(
                    server, int(gid))
            except IntegrityError:
                # Unique constraint violation from a concurrent
                # request — the record now exists.
                db.session.rollback()

            shared_server = SharedServer.query.filter_by(
                user_id=current_user.id,
                osid=server.id).first()

        if shared_server is None:
            raise Exception(
                "Failed to create shared server record "
                "for server {0}".format(server.id))

        return shared_server


class ServerMenuItem(MenuItem):
    def __init__(self, **kwargs):
        kwargs.setdefault("type", ServerModule.node_type)
        super().__init__(**kwargs)


blueprint = ServerModule(__name__)


class ServerNode(PGChildNodeView):
    node_type = ServerModule._NODE_TYPE
    node_label = "Server"

    parent_ids = [{'type': 'int', 'id': 'gid'}]
    ids = [{'type': 'int', 'id': 'sid'}]
    operations = dict({
        'obj': [
            {'get': 'properties', 'delete': 'delete', 'put': 'update'},
            {'get': 'list', 'post': 'create'}
        ],
        'nodes': [{'get': 'node'}, {'get': 'nodes'}],
        'sql': [{'get': 'sql'}],
        'msql': [{'get': 'modified_sql'}],
        'stats': [{'get': 'statistics'}],
        'dependency': [{'get': 'dependencies'}],
        'dependent': [{'get': 'dependents'}],
        'children': [{'get': 'children'}],
        'supported_servers.js': [{}, {}, {'get': 'supported_servers'}],
        'reload':
            [{'get': 'reload_configuration'}],
        'restore_point':
            [{'post': 'create_restore_point'}],
        'connect': [{
            'get': 'connect_status', 'post': 'connect', 'delete': 'disconnect'
        }],
        'change_password': [{'post': 'change_password'}],
        'wal_replay': [{
            'delete': 'pause_wal_replay', 'put': 'resume_wal_replay'
        }],
        'check_pgpass': [{'get': 'check_pgpass'}],
        'clear_saved_password': [{'put': 'clear_saved_password'}],
        'clear_sshtunnel_password': [{'put': 'clear_sshtunnel_password'}],
    })

    def update_connection_parameter(self, data, server, sharedserver=None):
        """
        This function is used to update the connection parameters.
        """
        if 'connection_params' in data and \
                hasattr(server, 'connection_params'):
            # For shared servers accessed by non-owners, apply changes
            # to the SharedServer's connection_params (a copy) so we
            # don't mutate the owner's Server record in-place.
            if sharedserver is not None and \
                    server.shared and \
                    server.user_id != current_user.id:
                existing_conn_params = dict(
                    sharedserver.connection_params or {})
            else:
                existing_conn_params = getattr(
                    server, 'connection_params')
            new_conn_params = data['connection_params']
            if 'deleted' in new_conn_params:
                for item in new_conn_params['deleted']:
                    if item['name'] in existing_conn_params:
                        del existing_conn_params[item['name']]
            if 'added' in new_conn_params:
                for item in new_conn_params['added']:
                    existing_conn_params[item['name']] = item['value']
            if 'changed' in new_conn_params:
                for item in new_conn_params['changed']:
                    existing_conn_params[item['name']] = item['value']

            data['connection_params'] = existing_conn_params

    @staticmethod
    def update_tags(data, server):
        """
        This function is used to update tags
        """
        old_tags = getattr(server, 'tags', [])
        # add old_text for comparison
        old_tags = [{**tag, 'old_text': tag['text']}
                    for tag in old_tags] if old_tags is not None else []
        new_tags_info = data.get('tags', None)

        def update_tag(tags, changed):
            for i, item in enumerate(tags):
                if item['old_text'] == changed['old_text']:
                    item = {**item, **changed}
                    tags[i] = item
                    break

        if new_tags_info:
            deleted_ids = [t['old_text']
                           for t in new_tags_info.get('deleted', [])]
            if len(deleted_ids) > 0:
                old_tags = [
                    t for t in old_tags if t['old_text'] not in deleted_ids
                ]

            for item in new_tags_info.get('changed', []):
                update_tag(old_tags, item)

            for item in new_tags_info.get('added', []):
                old_tags.append(item)

            # remove the old_text key
            data['tags'] = [
                {k: v for k, v in tag.items()
                 if k != 'old_text'} for tag in old_tags
            ]

    @pga_login_required
    def nodes(self, gid):
        res = []
        """
        Return a JSON document listing the servers under this server group
        for the user.
        """
        servers = get_user_server_query().filter(
            Server.servergroup_id == gid, Server.is_adhoc == 0)

        driver = get_driver(PG_DEFAULT_DRIVER)

        for server in servers:
            if _is_non_owner(server):
                shared_server = ServerModule.get_shared_server(server, gid)
                server = \
                    ServerModule.get_shared_server_properties(server,
                                                              shared_server)
            manager = driver.connection_manager(server.id)
            conn = manager.connection()
            connected = conn.connected()
            errmsg = None
            in_recovery = None
            wal_paused = None
            server_type = 'pg'
            if connected:
                server_type = manager.server_type
                status, result, in_recovery, wal_paused =\
                    recovery_state(conn, manager.version)
                if not status:
                    connected = False
                    manager.release()
                    errmsg = "{0} : {1}".format(server.name, result)

            res.append(
                self.blueprint.generate_browser_node(
                    "%d" % (server.id),
                    gid,
                    server.name,
                    server_icon_and_background(connected, manager, server),
                    True,
                    self.node_type,
                    connected=connected,
                    server_type=server_type,
                    version=manager.version,
                    db=manager.db,
                    host=server.host,
                    user=manager.user_info if connected else None,
                    in_recovery=in_recovery,
                    wal_pause=wal_paused,
                    is_password_saved=bool(server.save_password),
                    is_tunnel_password_saved=bool(server.tunnel_password),
                    errmsg=errmsg,
                    username=server.username,
                    shared=server.shared,
                    is_kerberos_conn=bool(server.kerberos_conn),
                    gss_authenticated=manager.gss_authenticated,
                    description=server.comment,
                    tags=server.tags
                )
            )

        if not len(res):
            return gone(errormsg=gettext(
                'The specified server group with id# {0} could not be found.'
            ))

        return make_json_response(result=res)

    @pga_login_required
    def node(self, gid, sid):
        """Return a JSON document listing the server groups for the user"""
        server = get_server(sid)

        if server is None:
            return make_json_response(
                status=410,
                success=0,
                errormsg=gettext(
                    "Could not find the server with id# {0}."
                ).format(sid)
            )

        if _is_non_owner(server):
            shared_server = ServerModule.get_shared_server(server, gid)
            server = ServerModule.get_shared_server_properties(server,
                                                               shared_server)

        manager = get_driver(PG_DEFAULT_DRIVER).connection_manager(server.id)
        conn = manager.connection()
        connected = conn.connected()
        errmsg = None
        in_recovery = None
        wal_paused = None
        if connected:
            try:
                status, result, in_recovery, wal_paused =\
                    recovery_state(conn, manager.version)

                if not status:
                    connected = False
                    manager.release()
                    errmsg = "{0} : {1}".format(server.name, result)

            except ConnectionLost:
                connected = False
                manager.release()

        return make_json_response(
            result=self.blueprint.generate_browser_node(
                "%d" % (server.id),
                gid,
                server.name,
                server_icon_and_background(connected, manager, server),
                True,
                self.node_type,
                connected=connected,
                server_type=manager.server_type if connected else 'pg',
                version=manager.version,
                db=manager.db,
                user=manager.user_info if connected else None,
                in_recovery=in_recovery,
                wal_pause=wal_paused,
                host=server.host,
                is_password_saved=bool(server.save_password),
                is_tunnel_password_saved=True
                if server.tunnel_password is not None else False,
                errmsg=errmsg,
                shared=server.shared,
                username=server.username,
                is_kerberos_conn=bool(server.kerberos_conn),
                gss_authenticated=manager.gss_authenticated,
                tags=server.tags
            ),
        )

    def delete_shared_server(self, gid, osid, user_id=None):
        """
        Delete SharedServer records for a given original server.
        :param gid: Server group ID
        :param osid: Original server ID
        :param user_id: If set, only delete for this user.
            If None, delete for ALL users (owner unshare/delete).
        """
        try:
            filters = dict(servergroup_id=gid, osid=osid)
            if user_id is not None:
                filters['user_id'] = user_id
            shared_server = SharedServer.query.filter_by(
                **filters)
            for s in shared_server:
                get_driver(PG_DEFAULT_DRIVER).delete_manager(s.id)
                db.session.delete(s)
            db.session.commit()

        except Exception as e:
            current_app.logger.exception(e)
            return make_json_response(
                success=0,
                errormsg=e.message)

    @pga_login_required
    def delete(self, gid, sid):
        """Delete a server node in the settings database."""
        servers = Server.query.filter_by(user_id=current_user.id, id=sid)
        server_name = None

        # TODO:: A server, which is connected, cannot be deleted
        if servers is None:
            return make_json_response(
                status=410,
                success=0,
                errormsg=gettext(
                    'The specified server could not be found.\n'
                    'Does the user have permission to access the '
                    'server?'
                )
            )
        else:
            try:
                for s in servers:
                    server_name = s.name
                    get_driver(PG_DEFAULT_DRIVER).delete_manager(s.id)
                    db.session.delete(s)
                db.session.commit()
                self.delete_shared_server(gid, sid)
                QueryHistory.clear_history(current_user.id, sid)

            except Exception as e:
                current_app.logger.exception(e)
                return make_json_response(
                    success=0,
                    errormsg=e.message)

        return make_json_response(success=1,
                                  info=gettext("Server deleted"))

    @permissions_required(AllPermissionTypes.object_register_server)
    @pga_login_required
    def update(self, gid, sid):
        """Update the server settings"""
        server = get_server(sid)
        sharedserver = None

        if server is None:
            return make_json_response(
                status=410,
                success=0,
                errormsg=gettext("Could not find the required server.")
            )

        if config.SERVER_MODE and server.shared and \
                server.user_id != current_user.id:
            sharedserver = ServerModule.get_shared_server(server, gid)

        # Not all parameters can be modified, while the server is connected
        config_param_map = {
            'name': 'name',
            'host': 'host',
            'port': 'port',
            'db': 'maintenance_db',
            'username': 'username',
            'password': 'password',
            'save_password': 'save_password',
            'gid': 'servergroup_id',
            'comment': 'comment',
            'role': 'role',
            'db_res': 'db_res',
            'db_res_type': 'db_res_type',
            'passexec_cmd': 'passexec_cmd',
            'passexec_expiration': 'passexec_expiration',
            'bgcolor': 'bgcolor',
            'fgcolor': 'fgcolor',
            'service': 'service',
            'use_ssh_tunnel': 'use_ssh_tunnel',
            'tunnel_host': 'tunnel_host',
            'tunnel_port': 'tunnel_port',
            'tunnel_username': 'tunnel_username',
            'tunnel_authentication': 'tunnel_authentication',
            'tunnel_identity_file': 'tunnel_identity_file',
            'tunnel_prompt_password': 'tunnel_prompt_password',
            'tunnel_keep_alive': 'tunnel_keep_alive',
            'shared': 'shared',
            'shared_username': 'shared_username',
            'kerberos_conn': 'kerberos_conn',
            'connection_params': 'connection_params',
            'prepare_threshold': 'prepare_threshold',
            'tags': 'tags',
            'post_connection_sql': 'post_connection_sql'
        }

        disp_lbl = {
            'name': gettext('name'),
            'port': gettext('Port'),
            'db': gettext('Maintenance database'),
            'username': gettext('Username'),
            'comment': gettext('Comments'),
            'role': gettext('Role')
        }

        data = request.form if request.form else json.loads(
            request.data
        )

        if 'db_res' in data and isinstance(data['db_res'], list):
            data['db_res'] = ','.join(data['db_res'])

        # Update connection parameter if any.
        self.update_connection_parameter(data, server, sharedserver)
        self.update_tags(data, sharedserver or server)

        if 'connection_params' in data and \
            'hostaddr' in data['connection_params'] and \
                not is_valid_ipaddress(data['connection_params']['hostaddr']):
            return make_json_response(
                success=0,
                status=400,
                errormsg=gettext('Not a valid Host address')
            )

        # remove the shared username if shared is updated to False
        if 'shared' in data and data['shared'] is False:
            data['shared_username'] = ''

        manager = get_driver(PG_DEFAULT_DRIVER).connection_manager(sid)
        conn = manager.connection()
        connected = conn.connected()

        self._server_modify_disallowed_when_connected(
            connected, data, disp_lbl)

        idx = self._set_valid_attr_value(gid, data, config_param_map, server,
                                         sharedserver)

        if idx == 0:
            return make_json_response(
                success=0,
                errormsg=gettext('No parameters were changed.')
            )

        # tags is JSON type, sqlalchemy sometimes will not detect change
        if 'tags' in data:
            flag_modified(sharedserver or server, 'tags')

        try:
            db.session.commit()
        except Exception as e:
            current_app.logger.exception(e)
            return make_json_response(
                success=0,
                errormsg=e.message
            )

        # When server is connected, we don't require to update the connection
        # manager. Because - we don't allow to change any of the parameters,
        # which will affect the connections.
        if not conn.connected():
            manager.update(server)
            # Suppress passexec for non-owners so the manager
            # never holds the owner's password-exec command.
            if _is_non_owner(server):
                manager.passexec = None

        return jsonify(
            node=self.blueprint.generate_browser_node(
                "%d" % (server.id), server.servergroup_id,
                server.name,
                server_icon_and_background(
                    connected, manager, sharedserver)
                if _is_non_owner(server)
                else server_icon_and_background(
                    connected, manager, server),
                True,
                self.node_type,
                connected=connected,
                shared=server.shared,
                user_id=server.user_id,
                user=manager.user_info if connected else None,
                server_type='pg',  # default server type
                username=server.username,
                role=server.role,
                is_password_saved=bool(server.save_password),
                description=server.comment,
                tags=(sharedserver or server).tags
            )
        )

    @staticmethod
    def _update_server_details(server, sharedserver,
                               config_param_map, arg, value):
        if value == '':
            value = None

        if _is_non_owner(server):
            setattr(sharedserver, config_param_map[arg], value)
        else:
            setattr(server, config_param_map[arg], value)

    def _set_valid_attr_value(self, gid, data, config_param_map, server,
                              sharedserver):

        idx = 0

        crypt_key_present, crypt_key = get_crypt_key()
        if not crypt_key_present:
            raise CryptKeyMissing

        # Fields that non-owners must never set on their
        # SharedServer — they enable command/SQL execution
        # or are owner-level concepts not on SharedServer.
        _owner_only_fields = frozenset({
            'passexec_cmd', 'passexec_expiration',
            'db_res', 'db_res_type',
        })

        for arg in config_param_map:
            if arg in data:
                # Non-owners cannot set dangerous fields.
                if _is_non_owner(server) and \
                        arg in _owner_only_fields:
                    continue
                value = data[arg]
                if arg == 'password':
                    value = encrypt(data[arg], crypt_key)
                # sqlite3 do not have boolean type so we need to
                # convert it manually to integer.
                # Only the owner may unshare — this deletes ALL
                # users' SharedServer records.
                if 'shared' in data and not data['shared'] \
                        and not _is_non_owner(server):
                    self.delete_shared_server(gid, server.id)
                if arg in ('sslcompression', 'use_ssh_tunnel',
                           'tunnel_authentication',
                           'kerberos_conn', 'shared'):
                    value = 1 if value else 0
                self._update_server_details(server, sharedserver,
                                            config_param_map, arg,
                                            value)
                idx += 1

        return idx

    def _server_modify_disallowed_when_connected(
            self, connected, data, disp_lbl):

        if connected:
            for arg in (
                    'db', 'role', 'service'
            ):
                if arg in data:
                    return forbidden(
                        errmsg=gettext(
                            "'{0}' cannot be modified "
                            "when the server is connected."
                        ).format(disp_lbl[arg])
                    )

    @pga_login_required
    def list(self, gid, object_filters):
        """
        Return list of attributes of all servers.
        """
        servers = get_user_server_query().filter(
            Server.servergroup_id == gid,
            Server.is_adhoc == 0).order_by(Server.name)
        sg = get_server_group(gid)
        res = []

        driver = get_driver(PG_DEFAULT_DRIVER)

        for server in servers:
            if _is_non_owner(server):
                shared_server = ServerModule.get_shared_server(server, gid)
                server = \
                    ServerModule.get_shared_server_properties(server,
                                                              shared_server)
            manager = driver.connection_manager(server.id)
            conn = manager.connection()
            connected = conn.connected()

            res.append({
                'id': server.id,
                'name': server.name,
                'host': server.host,
                'port': server.port,
                'db': server.maintenance_db,
                'username': server.username,
                'gid': server.servergroup_id,
                'group-name': sg.name,
                'comment': server.comment,
                'role': server.role,
                'connected': connected,
                'version': manager.ver,
                'server_type': manager.server_type if connected else 'pg',
                'db_res': get_db_restriction(server.db_res_type, server.db_res)
            })

        return ajax_response(
            response=res
        )

    @pga_login_required
    def properties(self, gid, sid):
        """Return list of attributes of a server"""

        server = get_server(sid)

        if server is None:
            return make_json_response(
                status=410,
                success=0,
                errormsg=self.not_found_error_msg()
            )
        server_owner = None
        sg = ServerGroup.query.filter_by(
            id=server.servergroup_id
        ).first()

        driver = get_driver(PG_DEFAULT_DRIVER)

        manager = driver.connection_manager(sid)
        conn = manager.connection()
        connected = conn.connected()

        # Get updated connection string to show on UI, if user change host,
        # port and user when server is connected
        display_connection_str = self.update_connection_string(manager, server)

        if _is_non_owner(server):
            shared_server = ServerModule.get_shared_server(server, gid)
            server = ServerModule.get_shared_server_properties(server,
                                                               shared_server)
            server_owner = server.server_owner

        use_ssh_tunnel = False
        tunnel_host = None
        tunnel_port = 22
        tunnel_username = None
        tunnel_authentication = False
        tunnel_keep_alive = 0
        connection_params = \
            convert_connection_parameter(server.connection_params)

        if server.use_ssh_tunnel:
            use_ssh_tunnel = bool(server.use_ssh_tunnel)
            tunnel_host = server.tunnel_host
            tunnel_port = server.tunnel_port
            tunnel_username = server.tunnel_username
            tunnel_authentication = bool(server.tunnel_authentication)
            tunnel_keep_alive = server.tunnel_keep_alive

        tags = None
        if server.tags is not None:
            tags = [{**tag, 'old_text': tag['text']}
                    for tag in server.tags]
        response = {
            'id': server.id,
            'name': server.name,
            'server_owner': server_owner,
            'user_id': server.user_id,
            'host': server.host,
            'port': server.port,
            'db': server.maintenance_db,
            'password': server.password,
            'save_password': server.save_password,
            'shared': server.shared if config.SERVER_MODE else None,
            'shared_username': server.shared_username
            if config.SERVER_MODE else None,
            'username': server.username,
            'gid': str(server.servergroup_id),
            'group-name': sg.name if (sg and sg.name) else gettext('Servers'),
            'comment': server.comment,
            'role': server.role,
            'connected': connected,
            'version': manager.ver,
            'server_type': manager.server_type if connected else 'pg',
            'bgcolor': server.bgcolor,
            'fgcolor': server.fgcolor,
            'db_res': get_db_restriction(server.db_res_type, server.db_res),
            'db_res_type': server.db_res_type,
            'passexec_cmd': server.passexec_cmd,
            'passexec_expiration': server.passexec_expiration,
            'service': server.service if server.service else None,
            'use_ssh_tunnel': use_ssh_tunnel,
            'tunnel_host': tunnel_host,
            'tunnel_port': tunnel_port,
            'tunnel_username': tunnel_username,
            'tunnel_identity_file': server.tunnel_identity_file
            if server.tunnel_identity_file else None,
            'tunnel_prompt_password': server.tunnel_prompt_password
            if server.tunnel_identity_file else 0,
            'tunnel_authentication': tunnel_authentication,
            'tunnel_keep_alive': tunnel_keep_alive,
            'kerberos_conn': bool(server.kerberos_conn),
            'gss_authenticated': manager.gss_authenticated,
            'gss_encrypted': manager.gss_encrypted,
            'cloud_status': server.cloud_status,
            'connection_params': connection_params,
            'connection_string': display_connection_str,
            'prepare_threshold': server.prepare_threshold,
            'tags': tags,
            'post_connection_sql': server.post_connection_sql,
        }

        return ajax_response(response)

    @staticmethod
    def update_connection_string(manager, server):
        # Get current connection info in dict.
        con_info = conninfo_to_dict(manager.display_connection_string)
        db_name = con_info['dbname'] if 'dbname' in con_info else None

        if 'host' in con_info and 'port' in con_info and 'user' in con_info:
            con_info.pop('host')
            con_info.pop('port')
            con_info.pop('user')

        # Create ordered dict to maintain the order of updated host, port,
        # dbname, user.
        con_info_ord = OrderedDict([('host', server.host),
                                    ('port', server.port),
                                    ('dbname', db_name),
                                    ('user', manager.user)])
        con_info_ord.update(con_info)
        display_conn_string = make_conninfo(**con_info_ord)
        return display_conn_string

    @permissions_required(AllPermissionTypes.object_register_server)
    @pga_login_required
    def create(self, gid):
        """Add a server node to the settings database"""
        required_args = ['name', 'db']

        data = request.form if request.form else json.loads(
            request.data
        )

        # Loop through data and if found any value is blank string then
        # convert it to None as after porting into React, from frontend
        # '' blank string is coming as a value instead of null.
        for item in data:
            if data[item] == '':
                data[item] = None

        # Get enc key
        crypt_key_present, crypt_key = get_crypt_key()
        if not crypt_key_present:
            raise CryptKeyMissing

        # Some fields can be provided with service file so they are optional
        if 'service' in data and not data['service']:
            required_args.extend([
                'host',
                'port',
                'username',
                'role'
            ])
        for arg in required_args:
            if arg not in data:
                return make_json_response(
                    status=410,
                    success=0,
                    errormsg=gettext(
                        "Could not find the required parameter ({})."
                    ).format(arg)
                )

        connection_params = convert_connection_parameter(
            data.get('connection_params', []))

        if 'hostaddr' in connection_params and \
                not is_valid_ipaddress(connection_params['hostaddr']):
            return make_json_response(
                success=0,
                status=400,
                errormsg=gettext('Not a valid Host address')
            )

        # To check ssl configuration
        _, connection_params = check_ssl_fields(connection_params)
        # set the connection params again in the data
        if 'connection_params' in data:
            data['connection_params'] = connection_params

        server = None
        db_restriction = None
        if 'db_res' in data and isinstance(data['db_res'], list):
            db_restriction = ','.join(data['db_res'])
        elif 'db_res' in data and 'db_res_type' in data and \
                data['db_res_type'] == RESTRICTION_TYPE_SQL:
            db_restriction = data['db_res']

        try:
            server = Server(
                user_id=current_user.id,
                servergroup_id=data.get('gid', gid),
                name=data.get('name'),
                host=data.get('host', None),
                port=data.get('port'),
                maintenance_db=data.get('db', None),
                username=data.get('username'),
                save_password=1 if data.get('save_password', False) and
                config.ALLOW_SAVE_PASSWORD else 0,
                comment=data.get('comment', None),
                role=data.get('role', None),
                db_res=db_restriction,
                db_res_type=data.get('db_res_type', None),
                bgcolor=data.get('bgcolor', None),
                fgcolor=data.get('fgcolor', None),
                service=data.get('service', None),
                use_ssh_tunnel=1 if data.get('use_ssh_tunnel', False) else 0,
                tunnel_host=data.get('tunnel_host', None),
                tunnel_port=data.get('tunnel_port', 22),
                tunnel_username=data.get('tunnel_username', None),
                tunnel_authentication=1 if data.get('tunnel_authentication',
                                                    False) else 0,
                tunnel_identity_file=data.get('tunnel_identity_file', None),
                tunnel_prompt_password=1 if data.get('tunnel_prompt_password',
                                                     True) else 0,
                tunnel_keep_alive=data.get('tunnel_keep_alive', 0),
                shared=data.get('shared', None),
                shared_username=data.get('shared_username', None),
                passexec_cmd=data.get('passexec_cmd', None),
                passexec_expiration=data.get('passexec_expiration', None),
                kerberos_conn=1 if data.get('kerberos_conn', False) else 0,
                connection_params=connection_params,
                prepare_threshold=data.get('prepare_threshold', None),
                tags=data.get('tags', None),
                post_connection_sql=data.get('post_connection_sql', None)
            )
            db.session.add(server)
            db.session.commit()
            connected = False
            user = None
            manager = None
            replication_type = None
            tunnel_password_saved = False

            if 'connect_now' in data and data['connect_now']:
                manager = get_driver(PG_DEFAULT_DRIVER).connection_manager(
                    server.id)
                manager.update(server)
                conn = manager.connection()

                have_password = False
                have_tunnel_password = False
                password = None
                passfile = None
                tunnel_password = ''
                if 'password' in data and data["password"] != '' and \
                        data["password"] is not None:
                    # login with password
                    have_password = True
                    password = data['password']
                    password = encrypt(password, crypt_key)
                elif 'passfile' in data['connection_params'] and \
                        data['connection_params']['passfile'] != '':
                    passfile = data['connection_params']['passfile']

                if 'tunnel_password' in data and data["tunnel_password"] != '':
                    have_tunnel_password = True
                    tunnel_password = data['tunnel_password']
                    tunnel_password = encrypt(tunnel_password, crypt_key)

                status, errmsg = conn.connect(
                    password=password,
                    passfile=passfile,
                    tunnel_password=tunnel_password,
                    server_types=ServerType.types()
                )
                if not status:
                    db.session.delete(server)
                    db.session.commit()
                    return make_json_response(
                        status=401,
                        success=0,
                        errormsg=gettext(
                            "Unable to connect to server:\n\n{}"
                        ).format(errmsg)
                    )
                else:
                    if 'save_password' in data and data['save_password'] and \
                            have_password and config.ALLOW_SAVE_PASSWORD:
                        setattr(server, 'password', password)
                        db.session.commit()

                    if 'save_tunnel_password' in data and \
                        data['save_tunnel_password'] and \
                        have_tunnel_password and \
                            config.ALLOW_SAVE_TUNNEL_PASSWORD:
                        setattr(server, 'tunnel_password', tunnel_password)
                        db.session.commit()

                    replication_type = get_replication_type(conn,
                                                            manager.version)
                    user = manager.user_info
                    connected = True

            return jsonify(
                node=self.blueprint.generate_browser_node(
                    "%d" % server.id, server.servergroup_id,
                    server.name,
                    server_icon_and_background(connected, manager, server),
                    True,
                    self.node_type,
                    username=server.username,
                    user=user,
                    connected=connected,
                    replication_type=replication_type,
                    shared=server.shared,
                    server_type=manager.server_type
                    if manager and manager.server_type
                    else 'pg',
                    version=manager.version
                    if manager and manager.version
                    else None,
                    is_kerberos_conn=bool(server.kerberos_conn),
                    gss_authenticated=manager.gss_authenticated if
                    manager and manager.gss_authenticated else False,
                    is_password_saved=bool(server.save_password),
                    is_tunnel_password_saved=tunnel_password_saved,
                    user_id=server.user_id,
                    tags=data.get('tags', None)
                )
            )

        except Exception as e:
            if server:
                db.session.delete(server)
                db.session.commit()

            current_app.logger.exception(e)
            return make_json_response(
                status=410,
                success=0,
                errormsg=str(e)
            )

    @pga_login_required
    def sql(self, gid, sid):
        return make_json_response(data='')

    @pga_login_required
    def modified_sql(self, gid, sid):
        return make_json_response(data='')

    @pga_login_required
    def statistics(self, gid, sid):
        manager = get_driver(PG_DEFAULT_DRIVER).connection_manager(sid)
        conn = manager.connection()

        if conn.connected():
            status, res = conn.execute_dict(
                render_template(
                    "/servers/sql/#{0}#/stats.sql".format(manager.version),
                    conn=conn, _=gettext
                )
            )

            if not status:
                return internal_server_error(errormsg=res)

            return make_json_response(data=res)

        return make_json_response(
            info=gettext(
                "Server has no active connection for generating statistics."
            )
        )

    @pga_login_required
    def dependencies(self, gid, sid):
        return make_json_response(data='')

    @pga_login_required
    def dependents(self, gid, sid):
        return make_json_response(data='')

    def supported_servers(self, **kwargs):
        """
        This property defines (if javascript) exists for this node.
        Override this property for your own logic.
        """

        return make_response(
            render_template(
                "servers/supported_servers.js",
                server_types=ServerType.types()
            ),
            200, {'Content-Type': MIMETYPE_APP_JS}
        )

    def connect_status(self, gid, sid):
        """Check and return the connection status."""
        server = get_server(sid)
        if server is None:
            return make_json_response(
                status=410, success=0,
                errormsg=self.not_found_error_msg()
            )
        manager = get_driver(PG_DEFAULT_DRIVER).connection_manager(sid)
        conn = manager.connection()
        connected = conn.connected()
        in_recovery = None
        wal_paused = None
        errmsg = None
        replication_type = None
        if connected:
            status, result, in_recovery, wal_paused =\
                recovery_state(conn, manager.version)

            if not status:
                connected = False
                manager.release()
                errmsg = "{0} : {1}".format(server.name, result)

            replication_type = get_replication_type(conn, manager.version)

        return make_json_response(
            data={
                'icon': server_icon_and_background(connected, manager, server),
                'connected': connected,
                'replication_type': replication_type,
                'in_recovery': in_recovery,
                'wal_pause': wal_paused,
                'server_type': manager.server_type if connected else "pg",
                'user': manager.user_info if connected else None,
                'errmsg': errmsg
            }
        )

    def is_prompt_tunnel_password(self, server):
        """
        This function will check whether to prompt tunnel password or not.
        """
        prompt_tunnel_password = True
        # In case of identity file check the value of tunnel_prompt_password.
        if server.tunnel_password is not None or \
            (server.tunnel_identity_file is not None and
             server.tunnel_prompt_password != 1):
            prompt_tunnel_password = False

        return prompt_tunnel_password

    def connect(self, gid, sid, is_qt=False, server=None):
        """
        Connect the Server and return the connection object.
        Verification Process before Connection:
            Verify requested server.

            Check the server password is already been stored in the
            database or not.
            If Yes, connect the server and return connection.
            If No, Raise HTTP error and ask for the password.

            In case of 'Save Password' request from user, encrypted Password
            will be stored in the respected server database and
            establish the connection OR just connect the server and do not
            store the password.
        """
        current_app.logger.info(
            'Connection Request for server#{0}'.format(sid)
        )

        # In case of Workspace layout ad-hoc server maybe pass to this
        # function in that case no need to fetch the server detail based on
        # sid.
        if server is None:
            server = get_server(sid)

        if server is None:
            return bad_request(self.not_found_error_msg())

        shared_server = None
        if _is_non_owner(server):
            shared_server = ServerModule.get_shared_server(server, gid)
            server = ServerModule.get_shared_server_properties(server,
                                                               shared_server)

        # Return if username is blank and the server is shared
        if server.username is None and not server.service and \
                server.shared:
            return make_json_response(
                status=200,
                success=0,
                errormsg=gettext(
                    "Please enter the server details to connect")
            )
        if current_user and hasattr(current_user, 'id'):
            # Fetch User Details.
            user = User.query.filter_by(id=current_user.id).first()
            if user is None:
                return unauthorized(gettext(UNAUTH_REQ))
        else:
            return unauthorized(gettext(UNAUTH_REQ))

        data = None
        if request.form:
            data = request.form
        elif request.data:
            data = json.loads(request.data)

        if data is None:
            data = {}

        password = None
        passfile = None
        tunnel_password = None
        save_password = False
        save_tunnel_password = False
        prompt_password = False
        prompt_tunnel_password = False

        # Connect the Server
        manager = get_driver(PG_DEFAULT_DRIVER).connection_manager(sid)
        # Update the manager with the server details if not connected and
        # the API call is not made from SQL Editor or View/Edit Data tool
        if not manager.connection().connected() and not is_qt:
            manager.update(server)
            # Re-suppress passexec after update() which rebuilds
            # from the (overlaid) server object.  Belt-and-suspenders:
            # the overlay already defaults passexec to None, but this
            # guards against direct DB edits.
            if _is_non_owner(server):
                manager.passexec = None
        conn = manager.connection()

        # Get enc key
        crypt_key_present, crypt_key = get_crypt_key()
        if not crypt_key_present:
            raise CryptKeyMissing

        # If server using SSH Tunnel
        if server.use_ssh_tunnel:
            if 'tunnel_password' not in data and \
                    server.tunnel_password is None:
                prompt_tunnel_password = self.is_prompt_tunnel_password(server)
            elif 'tunnel_password' not in data and \
                    server.tunnel_password is not None:
                tunnel_password = server.tunnel_password
            else:
                tunnel_password = data['tunnel_password'] \
                    if 'tunnel_password' in data else ''
                save_tunnel_password = data['save_tunnel_password'] \
                    if tunnel_password and 'save_tunnel_password' in data \
                    else False
                # Encrypt the password before saving with user's login
                # password key.
                try:
                    tunnel_password = encrypt(tunnel_password, crypt_key) \
                        if tunnel_password is not None else \
                        server.tunnel_password
                except Exception as e:
                    current_app.logger.exception(e)
                    return internal_server_error(errormsg=str(e))
        if 'password' not in data and (server.kerberos_conn is False or
                                       server.kerberos_conn is None):

            passfile_param = None
            if hasattr(server, 'connection_params') and \
                    'passfile' in server.connection_params:
                passfile_param = server.connection_params['passfile']

            conn_passwd = getattr(conn, 'password', None)
            if conn_passwd is None and not server.save_password and \
                    passfile_param is None and \
                    server.passexec_cmd is None and \
                    server.service is None:
                prompt_password = True
            elif passfile_param and passfile_param != '' and \
                    get_complete_file_path(passfile_param):
                passfile = passfile_param
            else:
                password = conn_passwd or server.password
        else:
            password = data['password'] if 'password' in data else None
            save_password = data['save_password']\
                if 'save_password' in data else False

            try:
                # Encrypt the password before saving with user's login
                # password key.
                password = encrypt(password, crypt_key) \
                    if password is not None else server.password
            except Exception as e:
                current_app.logger.exception(e)
                return internal_server_error(errormsg=str(e))

        # Check do we need to prompt for the database server or ssh tunnel
        # password or both. Return the password template in case password is
        # not provided, or password has not been saved earlier.
        if prompt_password or prompt_tunnel_password:
            return self.get_response_for_password(
                server, 428, prompt_password, prompt_tunnel_password)

        # Check whether to prompt for the tunnel password in case if
        # password is saved in server object or in data.
        prompt_tunnel_password = self.is_prompt_tunnel_password(server)

        try:
            status, errmsg = conn.connect(
                password=password,
                passfile=passfile,
                tunnel_password=tunnel_password,
                server_types=ServerType.types()
            )
        except Exception as e:
            return self.get_response_for_password(
                server, 401, not server.save_password, prompt_tunnel_password,
                getattr(e, 'message', str(e)))

        if not status:
            current_app.logger.error(
                "Could not connect to server(#{0}) - '{1}'.\nError: {2}"
                .format(server.id, server.name, errmsg)
            )
            if errmsg.find('Ticket expired') != -1:
                return internal_server_error(errmsg)

            return self.get_response_for_password(
                server, 401, not server.save_password,
                prompt_tunnel_password, errmsg)
        else:
            if save_password and config.ALLOW_SAVE_PASSWORD:
                try:
                    # 1 is True in SQLite as no boolean type
                    if _is_non_owner(server):
                        setattr(shared_server, 'save_password', 1)
                    else:
                        setattr(server, 'save_password', 1)

                    # Save the encrypted password using the user's login
                    # password key, if there is any password to save
                    if password:
                        if _is_non_owner(server):
                            setattr(shared_server, 'password', password)
                        else:
                            setattr(server, 'password', password)
                    db.session.commit()
                except Exception as e:
                    # Release Connection
                    current_app.logger.exception(e)
                    manager.release(database=server.maintenance_db)
                    conn = None

                    return internal_server_error(errormsg=e.message)

            if save_tunnel_password and config.ALLOW_SAVE_TUNNEL_PASSWORD:
                try:
                    # Save the encrypted tunnel password.
                    if _is_non_owner(server):
                        setattr(shared_server, 'tunnel_password',
                                tunnel_password)
                    else:
                        setattr(server, 'tunnel_password', tunnel_password)
                    db.session.commit()
                except Exception as e:
                    # Release Connection
                    current_app.logger.exception(e)
                    manager.release(database=server.maintenance_db)
                    conn = None

                    return internal_server_error(errormsg=e.message)

            current_app.logger.info('Connection Established for server: \
                %s - %s' % (server.id, server.name))
            # Update the recovery and wal pause option for the server
            # if connected successfully
            _, _, in_recovery, wal_paused =\
                recovery_state(conn, manager.version)

            replication_type = get_replication_type(conn, manager.version)

            return make_json_response(
                success=1,
                errormsg=errmsg,
                info=gettext("Server connected."),
                data={
                    "sid": server.id,
                    "did": manager.did,
                    'icon': server_icon_and_background(True, manager, server),
                    'connected': True,
                    'server_type': manager.server_type,
                    'replication_type': replication_type,
                    'type': manager.server_type,
                    'version': manager.version,
                    'db': manager.db,
                    'user': manager.user_info,
                    'in_recovery': in_recovery,
                    'wal_pause': wal_paused,
                    'is_password_saved': bool(server.save_password),
                    'is_tunnel_password_saved': True
                    if server.tunnel_password is not None else False,
                    'is_kerberos_conn': bool(server.kerberos_conn),
                    'gss_authenticated': manager.gss_authenticated
                }
            )

    def disconnect(self, gid, sid):
        """Disconnect the Server."""

        server = get_server(sid)
        if server is None:
            return bad_request(self.not_found_error_msg())

        # Release Connection
        manager = get_driver(PG_DEFAULT_DRIVER).connection_manager(sid)
        # Check if any psql terminal is running for the current disconnecting
        # server. If any terminate the psql tool connection.
        if 'sid_soid_mapping' in current_app.config and str(sid) in \
                current_app.config['sid_soid_mapping'] and \
                str(sid) in current_app.config['sid_soid_mapping']:
            for i in current_app.config['sid_soid_mapping'][str(sid)]:
                sio.emit('disconnect-psql', namespace='/pty', to=i)

        status = manager.release()

        if not status:
            return unauthorized(gettext("Server could not be disconnected."))
        else:
            return make_json_response(
                success=1,
                info=gettext("Server disconnected."),
                data={
                    'icon': server_icon_and_background(False, manager, server),
                    'connected': False
                }
            )

    def reload_configuration(self, gid, sid):
        """Reload the server configuration"""

        # Reload the server configurations
        manager = get_driver(PG_DEFAULT_DRIVER).connection_manager(sid)
        conn = manager.connection()

        if conn.connected():
            # Execute the command for reload configuration for the server
            status, _ = conn.execute_scalar("SELECT pg_reload_conf();")

            if not status:
                return internal_server_error(
                    gettext("Could not reload the server configuration.")
                )
            else:
                return make_json_response(data={
                    'status': True,
                    'result': gettext('Server configuration reloaded.')
                })

        else:
            return make_json_response(data={
                'status': False,
                'result': SERVER_CONNECTION_CLOSED})

    def create_restore_point(self, gid, sid):
        """
        This method will create named restore point

        Args:
            gid: Server group ID
            sid: Server ID

        Returns:
            None
        """
        try:
            data = None
            if request.form:
                data = request.form
            elif request.data:
                data = json.loads(request.data)
            restore_point_name = data['value'] if data else None
            manager = get_driver(PG_DEFAULT_DRIVER).connection_manager(sid)
            conn = manager.connection()

            # Execute SQL to create named restore point
            if conn.connected():
                if restore_point_name:
                    status, res = conn.execute_scalar(
                        "SELECT pg_create_restore_point('{0}');".format(
                            restore_point_name
                        )
                    )
                    if not status:
                        return internal_server_error(
                            errormsg=str(res)
                        )

                return make_json_response(
                    data={
                        'status': 1,
                        'result': gettext(
                            'Named restore point created: {0}').format(
                                restore_point_name)
                    })

        except Exception as e:
            current_app.logger.error(gettext(
                'Named restore point creation failed ({0})').format(
                    str(e))
            )
            return internal_server_error(errormsg=str(e))

    def change_password(self, gid, sid):
        """
        This function is used to change the password of the
        Database Server.

        Args:
            gid: Group id
            sid: Server id
        """
        try:
            data = None
            if request.form:
                data = request.form
            elif request.data:
                data = json.loads(request.data)

            # Get enc key
            crypt_key_present, crypt_key = get_crypt_key()
            if not crypt_key_present:
                raise CryptKeyMissing

            # Fetch Server Details
            server = get_server(sid, only_owned=False)
            if server is None:
                return bad_request(self.not_found_error_msg())

            # Fetch User Details.
            user = User.query.filter_by(id=current_user.id).first()
            if user is None:
                return unauthorized(gettext(UNAUTH_REQ))

            manager = get_driver(PG_DEFAULT_DRIVER).connection_manager(sid)
            conn = manager.connection()
            is_passfile = False

            # If there is no password found for the server
            # then check for pgpass file
            if not server.password and not manager.password and \
                hasattr(server, 'connection_params') and \
                'passfile' in server.connection_params and \
                manager.get_connection_param_value('passfile') and \
                server.connection_params['passfile'] == \
                    manager.get_connection_param_value('passfile'):
                is_passfile = True

            # Check for password only if there is no pgpass file used
            if not is_passfile and data and \
                    ('password' not in data or data['password'] == ''):
                return make_json_response(
                    status=400,
                    success=0,
                    errormsg=gettext(
                        "Could not find the required parameter(s)."
                    )
                )

            if data and ('newPassword' not in data or
                         data['newPassword'] == '' or
                         'confirmPassword' not in data or
                         data['confirmPassword'] == ''):
                return make_json_response(
                    status=400,
                    success=0,
                    errormsg=gettext(
                        "Could not find the required parameter(s)."
                    )
                )

            if data['newPassword'] != data['confirmPassword']:
                return make_json_response(
                    status=200,
                    success=0,
                    errormsg=gettext(
                        "Passwords do not match."
                    )
                )

            # Check against old password only if no pgpass file
            if not is_passfile:
                decrypted_password = decrypt(manager.password, crypt_key)
                if isinstance(decrypted_password, bytes):
                    decrypted_password = decrypted_password.decode()
                password = data['password']

                # Validate old password before setting new.
                if password != decrypted_password:
                    return unauthorized(gettext("Incorrect password."))

            # Hash new password before saving it.
            password = conn.pq_encrypt_password_conn(data['newPassword'],
                                                     manager.user)
            if password is None:
                return internal_server_error(errormsg="Unable to"
                                                      " change the password.")

            SQL = render_template(
                "/servers/sql/#{0}#/change_password.sql".format(
                    manager.version),
                conn=conn, _=gettext,
                user=manager.user, encrypted_password=password)

            status, res = conn.execute_scalar(SQL)

            if not status:
                return internal_server_error(errormsg=res)

            # Store password in sqlite only if no pgpass file
            if not is_passfile:
                password = encrypt(data['newPassword'], crypt_key)
                # Check if old password was stored in pgadmin4
                # sqlite database. If yes then update that password.
                # For non-owners of shared servers, check the
                # SharedServer record (not the owner's Server).
                if config.ALLOW_SAVE_PASSWORD:
                    if server.shared and \
                            server.user_id != current_user.id:
                        shared_server = \
                            ServerModule.get_shared_server(
                                server, gid)
                        if shared_server and \
                                shared_server.password is not None:
                            setattr(shared_server, 'password',
                                    password)
                            db.session.commit()
                    elif server.password is not None:
                        setattr(server, 'password', password)
                        db.session.commit()
                # Also update password in connection manager.
                manager.password = password
                manager.update_session()

            return make_json_response(
                status=200,
                success=1,
                info=gettext(
                    "Password changed successfully."
                )
            )

        except Exception as e:
            return internal_server_error(errormsg=str(e))

    def wal_replay(self, sid, pause=True):
        """
        Utility function for wal_replay for resume/pause.
        """
        server = get_server(sid)

        if server is None:
            return make_json_response(
                success=0,
                errormsg=self.not_found_error_msg()
            )

        try:
            manager = get_driver(PG_DEFAULT_DRIVER).connection_manager(sid)
            conn = manager.connection()
            msg = None
            # Execute SQL to pause or resume WAL replay
            if conn.connected():
                if pause:
                    sql = "SELECT pg_xlog_replay_pause();"
                    if manager.version >= 100000:
                        sql = "SELECT pg_wal_replay_pause();"

                    status, res = conn.execute_scalar(sql)
                    if not status:
                        return internal_server_error(
                            errormsg=str(res)
                        )
                    msg = gettext('WAL replay paused')
                else:
                    sql = "SELECT pg_xlog_replay_resume();"
                    if manager.version >= 100000:
                        sql = "SELECT pg_wal_replay_resume();"

                    status, res = conn.execute_scalar(sql)
                    if not status:
                        return internal_server_error(
                            errormsg=str(res)
                        )
                    msg = gettext('WAL replay resumed')
                return make_json_response(
                    success=1,
                    info=msg,
                    data={'in_recovery': True, 'wal_pause': pause}
                )
            return gone(errormsg=gettext('Please connect to the server.'))
        except Exception as e:
            current_app.logger.error(
                'WAL replay pause/resume failed'
            )
            return internal_server_error(errormsg=str(e))

    def resume_wal_replay(self, gid, sid):
        """
        This method will resume WAL replay

        Args:
            gid: Server group ID
            sid: Server ID

        Returns:
            None
        """
        return self.wal_replay(sid, False)

    def pause_wal_replay(self, gid, sid):
        """
        This method will pause WAL replay

        Args:
            gid: Server group ID
            sid: Server ID

        Returns:
            None
        """
        return self.wal_replay(sid, True)

    def check_pgpass(self, gid, sid):
        """
        This function is used to check whether server is connected
        using pgpass file or not

        Args:
            gid: Group id
            sid: Server id
        """
        is_pgpass = False
        server = get_server(sid)

        if server is None:
            return make_json_response(
                success=0,
                errormsg=self.not_found_error_msg()
            )

        try:
            manager = get_driver(PG_DEFAULT_DRIVER).connection_manager(sid)
            conn = manager.connection()
            if not conn.connected():
                return gone(
                    errormsg=gettext('Please connect to the server.')
                )

            if (not server.password or not manager.password) and \
                hasattr(server, 'connection_params') and \
                'passfile' in server.connection_params and \
                manager.get_connection_param_value('passfile') and \
                server.connection_params['passfile'] == \
                    manager.get_connection_param_value('passfile'):
                is_pgpass = True
            return make_json_response(
                success=1,
                data=dict({'is_pgpass': is_pgpass}),
            )
        except Exception as e:
            current_app.logger.error(
                'Unable to fetch pgpass status'
            )
            return internal_server_error(errormsg=str(e))

    def get_response_for_password(self, server, status, prompt_password=False,
                                  prompt_tunnel_password=False, errmsg=None):

        if server.use_ssh_tunnel:
            data = {
                "server_label": server.name,
                "username": server.username,
                "tunnel_username": server.tunnel_username,
                "tunnel_host": server.tunnel_host,
                "tunnel_identity_file": server.tunnel_identity_file,
                "tunnel_keep_alive": server.tunnel_keep_alive,
                "errmsg": errmsg,
                "service": server.service,
                "prompt_tunnel_password": prompt_tunnel_password,
                "prompt_password": prompt_password,
                "allow_save_password":
                    True if config.ALLOW_SAVE_PASSWORD and
                    'allow_save_password' in session and
                    session['allow_save_password'] else False,
                "allow_save_tunnel_password":
                    True if config.ALLOW_SAVE_TUNNEL_PASSWORD and
                    'allow_save_password' in session and
                    session['allow_save_password'] else False
            }
            return make_json_response(
                success=0,
                status=status,
                result=data
            )
        else:
            data = {
                "sid": server.id,
                "server_label": server.name,
                "username": server.username,
                "errmsg": errmsg,
                "service": server.service,
                "prompt_password": True,
                "allow_save_password":
                    True if config.ALLOW_SAVE_PASSWORD and
                    'allow_save_password' in session and
                    session['allow_save_password'] else False
            }
            return make_json_response(
                success=0,
                status=status,
                result=data
            )

    def clear_saved_password(self, gid, sid):
        """
        This function is used to remove database server password stored into
        the pgAdmin's db file.

        :param gid:
        :param sid:
        :return:
        """
        try:
            server = get_server(sid, only_owned=False)
            if server is None:
                return make_json_response(
                    success=0,
                    info=self.not_found_error_msg()
                )

            if _is_non_owner(server):
                shared_server = ServerModule.get_shared_server(
                    server, gid)
                setattr(shared_server, 'password', None)
                if shared_server.save_password:
                    setattr(shared_server, 'save_password', 0)
            else:
                setattr(server, 'password', None)
                if server.save_password:
                    setattr(server, 'save_password', 0)
            db.session.commit()
        except Exception as e:
            current_app.logger.error(
                "Unable to clear saved password.\nError: {0}".format(str(e))
            )

            return internal_server_error(errormsg=str(e))

        return make_json_response(
            success=1,
            info=gettext("The saved password was cleared successfully."),
            data={'is_password_saved': False}
        )

    def clear_sshtunnel_password(self, gid, sid):
        """
        This function is used to remove sshtunnel password stored into
        the pgAdmin's db file.

        :param gid:
        :param sid:
        :return:
        """
        try:
            server = get_server(sid, only_owned=False)
            if server is None:
                return make_json_response(
                    success=0,
                    info=self.not_found_error_msg()
                )

            if _is_non_owner(server):
                shared_server = ServerModule.get_shared_server(
                    server, gid)
                setattr(shared_server, 'tunnel_password', None)
            else:
                setattr(server, 'tunnel_password', None)
            db.session.commit()
        except Exception as e:
            current_app.logger.error(
                "Unable to clear ssh tunnel password."
                "\nError: {0}".format(str(e))
            )

            return internal_server_error(errormsg=str(e))

        return make_json_response(
            success=1,
            info=gettext("The saved password was cleared successfully."),
            data={'is_tunnel_password_saved': False}
        )


SchemaDiffRegistry(blueprint.node_type, ServerNode)
ServerNode.register_node_view(blueprint)