Source code for schrodinger.application.jobserver_cert_gui_dir.registration_dialogs
import os
from . import jobserver_cert_gui_icons_rc # noqa: F401
from schrodinger.application.jobserver_cert_gui_dir import registration_panel_ui
from schrodinger.infra import mmjob
from schrodinger.job import cert
from schrodinger.job import server
from schrodinger.models import parameters
from schrodinger.Qt import QtGui
from schrodinger.Qt import QtWidgets
from schrodinger.Qt.QtWidgets import QLineEdit
from schrodinger.ui.qt import basewidgets
from schrodinger.ui.qt.appframework2 import settings
from schrodinger.ui.qt.decorators import wait_cursor
from schrodinger.ui.qt.standard.icons import icons
STATUS_TEXT_FORMAT = """<html style="color:{color}">{status}</html>"""
REGISTERED_GREEN_COLOR = "#339900"
INVALID_RED_COLOR = "#CC0000"
UNREGISTERED_ORANGE_COLOR = "#c87c00"
UNAVAILABLE_GRAY_COLOR = "#666666"
CONFIGURED_SERVER_REGISTER_TOOLTIP = "Register with selected server"
NEW_SERVER_REGISTER_TOOLTIP = "Configure server and register with it"
DO_NOT_SHOW_REGISTRATION_PANEL = "DO_NOT_SHOW_REG_PANEL"
[docs]class ServerRegistrationMixin:
"""
A mixin for providing shared logic for server registration GUIs.
Subclasses must also inherit from `mappers.MapperMixin`.
"""
model_class = RegistrationInput
[docs] def defineMappings(self):
M = self.model_class
ui = self.ui
return [
(ui.username_value_le, M.username),
(ui.pswd_value_le, M.password)
] # yapf: disable
[docs] def getSignalsAndSlots(self, model):
return [
(model.usernameChanged, self.onUsernameOrPasswordChanged),
(model.passwordChanged, self.onUsernameOrPasswordChanged)
] # yapf: disable
[docs] def onUsernameOrPasswordChanged(self):
raise NotImplementedError
[docs]class RegistrationPanel(ServerRegistrationMixin, basewidgets.Panel):
"""
A panel to register for the certification of job server on a host address.
The panel allows user to choose from already configured set of server
addresses or register for a new server address as well.
"""
ui_module = registration_panel_ui
[docs] def initSetOptions(self):
super().initSetOptions()
self.std_btn_specs = {self.StdBtn.Cancel: (None, 'Close')}
self.help_topic = 'JOB_SERVER_REGISTRATION_PANEL'
[docs] def initSetUp(self):
super().initSetUp()
self.setWindowTitle("Register with Job Server")
self.ui.pswd_value_le.setEchoMode(QLineEdit.Password)
self.registered_servers, self.unregistered_servers, self.invalid_registration_servers,\
self.unavailable_servers = set(), set(), set(), set()
self.user_registered_servers = set()
# Cache the localhost dir as the SCHRODINGER_LOCALHOST_JOBSERVER_DIRECTORY
# environment variable to improve performance of `cert.verify_cert`
os.environ[
"SCHRODINGER_LOCALHOST_JOBSERVER_DIRECTORY"] = mmjob.get_localhost_jobserver_dir(
)
self._do_not_show_cb = QtWidgets.QCheckBox(
parent=self,
text="Do not show again (not using any remote servers)")
self.bottom_left_layout.addWidget(self._do_not_show_cb)
if settings.get_persistent_value(DO_NOT_SHOW_REGISTRATION_PANEL, ''):
self._do_not_show_cb.setChecked(True)
self._do_not_show_cb.toggled.connect(self._onDoNotShowCheckboxToggled)
self.categorizeServers()
self.initServersCombo()
# Hide the `do_not_show checkbox` if the user has registered job server with
# at least one remote host
if self.registered_servers and not (
len(self.registered_servers) == 1 and
list(self.registered_servers)[0].startswith("localhost")):
self._do_not_show_cb.setVisible(False)
self.ui.configured_server_rb.toggled.connect(
self._updateServerInfoWidgets)
self.ui.configured_server_rb.setChecked(True)
self.ui.register_btn.clicked.connect(self.onRegisterClicked)
self.ui.server_names_combo.currentIndexChanged.connect(
self._updateStatusInfo)
self.ui.remove_server_btn.clicked.connect(self._onRemoveServerClicked)
self._updateServerInfoWidgets()
def _onDoNotShowCheckboxToggled(self, toggled):
"""
Store the checked value of `do_not_show checkbox` persistently when
toggled.
:param toggled: Whether the checkbox is toggled on.
:type toggled: bool
"""
settings.set_persistent_value(DO_NOT_SHOW_REGISTRATION_PANEL, toggled)
[docs] def hideDoNotShowCheckbox(self):
self._do_not_show_cb.setVisible(False)
def _onRemoveServerClicked(self):
"""
Remove the certification for user configured servers as per the response
from user to question boxes.
"""
curr_server_address = self.getCurrServerAddress()
if curr_server_address in self.registered_servers:
ok = self.question(
f'<p>Remove Job Server {curr_server_address} from the'
f' list of configured servers? This will delete your'
f' registration information, and you will no longer '
f'be able to run jobs on the associated hosts.</p>'
f'<p>Restart Maestro to complete this action.</p>',
title='Confirm Removal')
if not ok:
return
else:
ok = self.question(
f'<p>Remove Job Server {curr_server_address}'
f' from the list of configured servers?</p>'
f'<p>Restart Maestro to complete this action.</p>',
title='Confirm Removal')
if not ok:
return
try:
cert.remove_cert(curr_server_address)
except RuntimeError:
self.error("Could not remove server certification.")
else:
self.unavailable_servers.discard(curr_server_address)
self.registered_servers.discard(curr_server_address)
self.invalid_registration_servers.discard(curr_server_address)
self.unregistered_servers.discard(curr_server_address)
self.user_registered_servers.discard(curr_server_address)
self.initServersCombo()
def _updateServerInfoWidgets(self):
"""
Make the port info for servers visible if user switches to new server
registration. Show the status of current server selected out of
configured servers.
"""
is_server_configured = self.ui.configured_server_rb.isChecked()
self.ui.port_lbl.setVisible(not is_server_configured)
self.ui.port_value_le.setVisible(not is_server_configured)
self.ui.new_server_le.setVisible(not is_server_configured)
self.ui.icon_lbl.setVisible(is_server_configured)
self.ui.status_lbl.setVisible(is_server_configured)
self.ui.server_names_combo.setVisible(is_server_configured)
self.ui.remove_server_btn.setVisible(is_server_configured)
self.ui.register_btn.setToolTip(
CONFIGURED_SERVER_REGISTER_TOOLTIP
if is_server_configured else NEW_SERVER_REGISTER_TOOLTIP)
self._updateStatusInfo()
def _updateStatusInfo(self):
"""
Update the status icon and text as per whether the current item in
combobox is registered, unregistered, invalid registration or unavailable.
"""
curr_server_address = self.getCurrServerAddress()
if curr_server_address in self.registered_servers:
pixmap = QtGui.QPixmap(icons.OK_LB)
status_text = STATUS_TEXT_FORMAT.format(
color=REGISTERED_GREEN_COLOR, status='Registered')
elif curr_server_address in self.unregistered_servers:
pixmap = QtGui.QPixmap(icons.VALIDATION_WARNING_LB)
status_text = STATUS_TEXT_FORMAT.format(
color=UNREGISTERED_ORANGE_COLOR, status='Not registered')
elif curr_server_address in self.invalid_registration_servers:
pixmap = QtGui.QPixmap(icons.VALIDATION_ERROR_LB)
status_text = STATUS_TEXT_FORMAT.format(
color=INVALID_RED_COLOR, status='Registration invalid')
else:
pixmap = QtGui.QPixmap(
":/jobserver_cert_gui_dir/icons/unavailable.png")
status_text = STATUS_TEXT_FORMAT.format(
color=UNAVAILABLE_GRAY_COLOR, status='Unavailable')
pixmap = pixmap.scaledToHeight(30)
self.ui.icon_lbl.setPixmap(pixmap)
self.ui.status_lbl.setText(status_text)
self.ui.remove_server_btn.setEnabled(
curr_server_address in self.user_registered_servers)
[docs] def onUsernameOrPasswordChanged(self):
"""
Enable the register button only when both username and password
have text filled.
"""
enable = bool(self.model.username) and bool(self.model.password)
self.ui.register_btn.setEnabled(enable)
[docs] def categorizeServers(self):
"""
Categorize all registered servers into the four categories of registered,
invalid registration, unregistered and unavailable.
"""
self.user_registered_servers = mmjob.get_registered_jobservers()
all_servers = self.user_registered_servers.union(
cert.configured_servers())
schrodinger = os.environ["SCHRODINGER"]
for server_address in all_servers:
try:
server.get_server_info(schrodinger, server_address)
except RuntimeError:
self.unavailable_servers.add(server_address)
available_servers = all_servers - self.unavailable_servers
servers_without_registration = cert.servers_without_registration()
self.unregistered_servers = servers_without_registration - self.unavailable_servers
servers_with_registration = available_servers - servers_without_registration
for server_address in servers_with_registration:
try:
cert.verify_cert(server_address)
except RuntimeError:
self.invalid_registration_servers.add(server_address)
self.registered_servers = servers_with_registration - self.invalid_registration_servers
[docs] def initServersCombo(self):
"""
Initialize the servers combobox to contain the servers in the order of invalid,
unavailable, unregistered and registered.
"""
self.ui.server_names_combo.clear()
registration_server_combo_list = list(
self.invalid_registration_servers) + list(
self.unavailable_servers) + list(
self.unregistered_servers) + list(self.registered_servers)
self.ui.server_names_combo.addItems(registration_server_combo_list)
[docs] def onRegisterClicked(self):
"""
Register the user with the provided credentials. Show the appropriate dialog box
as per whether registration was successful or not.
"""
is_server_configured = self.ui.configured_server_rb.isChecked()
schrodinger = os.environ["SCHRODINGER"]
curr_server_address = self.getCurrServerAddress()
if not is_server_configured:
try:
server.get_server_info(schrodinger, curr_server_address)
except RuntimeError:
self.error(
"No Job Server could be found at the specified location. Check the"
" machine name and port number")
return
try:
self.register()
except (cert.BadLDAPInputException, cert.AuthenticationException,
RuntimeError) as error:
self.error("Authentication failed.")
if not is_server_configured:
self.unregistered_servers.add(curr_server_address)
self.initServersCombo()
self.ui.configured_server_rb.setChecked(True)
self.ui.server_names_combo.setCurrentText(curr_server_address)
else:
self.info(
"Registration was successful. Restart Maestro to complete the action.",
title="Registration Successful")
self.unregistered_servers.discard(curr_server_address)
self.invalid_registration_servers.discard(curr_server_address)
self.unavailable_servers.discard(curr_server_address)
self.registered_servers.add(curr_server_address)
if not is_server_configured:
self.user_registered_servers.add(curr_server_address)
self.initServersCombo()
# Disable and uncheck the checkbox if registration succeeded
self._do_not_show_cb.setChecked(False)
self._do_not_show_cb.setEnabled(False)
[docs] @wait_cursor
def register(self):
"""
Register with a certificate for the current server address and user credentials.
See `cert.get_cert` for raised exceptions.
"""
hostname, port = cert.hostname_and_port(self.getCurrServerAddress())
cert.get_cert(hostname,
port,
self.model.username,
ssh_password=self.model.password,
ldap_password=self.model.password)
[docs] def getCurrServerAddress(self):
"""
Construct and return the current server address.
:rtype: string
"""
is_server_configured = self.ui.configured_server_rb.isChecked()
if is_server_configured:
return self.ui.server_names_combo.currentText()
else:
if not self.ui.port_value_le.text():
return self.ui.new_server_le.text()
return f'{self.ui.new_server_le.text()}:{self.ui.port_value_le.text()}'