🌱🏠🖥️ Cross-platform desktop application for greenhouse https://greenhouse.server.garden/
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 

1155 lines
43 KiB

from fbs_runtime.application_context.PyQt5 import ApplicationContext
from PyQt5.QtCore import Qt, QThreadPool, QRect
from PyQt5.QtGui import QPixmap, QColor, QIcon, QPalette
from PyQt5.QtWidgets import QWidget, QDialog, QLabel, QLineEdit, QPushButton, QFrame, QVBoxLayout, QHBoxLayout, QMessageBox, QScrollArea, QComboBox, QCheckBox, QSizePolicy
from qtwidgets import PasswordEdit
from pyqtspinner.spinner import WaitingSpinner
from typing import Callable, Tuple, Dict
import pyqtspinner
import socket
import requests
import urllib3
import time
import sys
import sip
from worker import Worker
# remove unverified TLS warning
# https://urllib3.readthedocs.io/en/latest/advanced-usage.html#ssl-warnings
urllib3.disable_warnings()
daemon_api_port = 9572
greenhouse_cloud_url = "http://localhost:8081"
def my_exec_info_message(exec_info):
return "{}: {}".format(".".join([exec_info[0].__module__, exec_info[0].__name__]), exec_info[1])
class MainWindow(QWidget):
def __init__(self, images: Dict[str, str]):
super().__init__()
self.icon_create = images["icon_create"]
self.icon_open_socket = images["icon_open_socket"]
self.icon_open_folder = images["icon_open_folder"]
self.icon_success = images["icon_success"]
self.icon_failure = images["icon_failure"]
self.image_tunnel_arrow = images["image_tunnel_arrow"]
self.register_button = None
self.apply_config_button = None
self.status = None
self.underlying_services_frame = None
self.apply_config_dialog = None
self.apply_config_dialog_content = None
self.last_apply_config_status_index = None
self.last_apply_config_status_error = None
self.status_polling = False
self.tunnels = []
self.screens = dict(
init= self.create_initial_loading_frame(),
login= self.create_login_frame(images["logo"]),
main= self.create_main_frame(),
)
self.screens_parent = QVBoxLayout()
for kv in self.screens.items():
self.screens_parent.addWidget(kv[1], 2)
self.setLayout(self.screens_parent)
self.set_active_screen("init")
self.resize(900, 700)
self.get_status(self.get_status_callback, update_tenant_info=True, must_succeed=False)
def set_active_screen(self, name: str):
for kv in self.screens.items():
if kv[0] == name:
kv[1].show()
else:
kv[1].hide()
def get_status(self, callback: Callable[[Tuple[dict,str]], None], update_tenant_info=False, must_succeed=False):
get_status_worker = Worker(self.status_task, update_tenant_info, must_succeed)
get_status_worker.signals.result.connect(callback)
QThreadPool.globalInstance().start(get_status_worker)
def status_task(self, update_tenant_info: bool, must_succeed: bool) -> Tuple[dict,str]:
try:
status_url = f'https://127.0.0.1:{daemon_api_port}/status{"?updateTenantInfo=true" if update_tenant_info else ""}'
result = requests.post( status_url, verify=False, timeout=5)
if not result.ok:
if must_succeed:
return (None, f"status request failed, server returned HTTP {result.status_code}: {result.text}")
else:
return (None, None)
result_json = result.json()
#print(result_json)
update_tenant_info_message = result_json['update_tenant_info_message']
if must_succeed and update_tenant_info_message != "success":
return (None, f"get account from greenhouse failed: {update_tenant_info_message}")
return (result_json, None)
except:
if must_succeed:
return (None, "status request failed: " + my_exec_info_message(sys.exc_info()))
else:
return (None, None)
def get_status_callback(self, result: Tuple[dict,str,bool]) -> None:
status = result[0]
error = result[1]
if error is not None and error is not "":
self.error_popup("Background Service Error", "couldn't get current status from the greenhouse background service:\n\n" + error)
else:
if status is None:
# if status and error are both None that means must_succeed was False:
# the user is probably not registered yet.
self.set_active_screen("login")
elif type(status) is dict and "needs_api_token" in status:
if status["needs_api_token"]:
self.set_active_screen("login")
else:
#print("status:", status)
self.status = status
self.tunnels = status['tunnels'] if 'tunnels' in status and status['tunnels'] is not None else []
self.update_main_screen()
self.set_active_screen("main")
self.start_status_polling()
update_tenant_info_message = "update_tenant_info_message was missing"
if "update_tenant_info_message" in status:
update_tenant_info_message = status["update_tenant_info_message"]
if update_tenant_info_message != "n/a" and update_tenant_info_message != "success":
self.error_popup("Background Service Error", f"couldn't get current status of your greenhouse account because: {update_tenant_info_message}\n\nThe greenhouse cloud service may be experiencing an outage and the application may not work.")
else:
self.error_popup("Internal Error", f"expected a dictionary containing the key 'needs_api_token', instead got {status}")
def start_status_polling(self):
if not self.status_polling:
self.status_polling = True
status_polling_worker = Worker(self.status_polling_task)
QThreadPool.globalInstance().start(status_polling_worker)
def status_polling_task(self):
while True:
time.sleep(0.5)
self.get_status(self.get_status_polling_callback, update_tenant_info=False, must_succeed=False)
def get_status_polling_callback(self, result: Tuple[dict,str,bool]) -> None:
status = result[0]
error = result[1]
if error is not None:
return
self.status = status
if self.apply_config_dialog is not None:
self.update_apply_config_dialog()
else:
self.update_underlying_services_status()
def update_main_screen(self):
# how to remove widget https://stackoverflow.com/questions/5899826/pyqt-how-to-remove-a-widget
if "main" in self.screens and self.screens["main"] is not None:
self.screens_parent.removeWidget(self.screens["main"])
sip.delete(self.screens["main"])
self.screens["main"] = None
self.apply_config_button = None
self.screens["main"] = self.create_main_frame()
self.screens_parent.addWidget(self.screens["main"])
def update_tunnels_list(self):
# how toremove widget
if self.gui_tunnels_list is not None:
self.tunnels_frame_layout.removeWidget(self.gui_tunnels_list)
sip.delete(self.gui_tunnels_list)
self.gui_tunnels_list = None
self.gui_tunnels_list = self.create_tunnels_list()
self.tunnels_frame_layout.addWidget(self.gui_tunnels_list)
self.tunnels_frame_layout.setAlignment(self.gui_tunnels_list, Qt.AlignTop)
def create_tunnel(self):
self.tunnels.append(dict())
self.update_tunnels_list()
def register_clicked(self, hostname: str, api_token: str):
self.register_button.setDisabled(True)
register_worker = Worker(self.register_task, hostname, api_token)
register_worker.signals.result.connect(self.register_callback)
QThreadPool.globalInstance().start(register_worker)
def register_task(self, hostname: str, api_token: str) -> Tuple[str,str,str]:
try:
# ask greenhouse is this user already has a server connected with this servername
# TODO real greenhouse
result = requests.post(
f'{greenhouse_cloud_url}/api/tenant_info',
headers={'Authorization': f'Bearer {api_token}'},
timeout=8
)
if not result.ok:
return (None, None, f"tenant info request failed, server returned HTTP {result.status_code}: {result.text}")
result_json = result.json()
client_states = result_json['ClientStates']
if hostname in client_states and client_states[hostname] == "ClientConnected":
return (None, None, f"Error: Can't register the server name '{hostname}' because a server with that name is already connected.")
# call the greenhouse daemon to do the work of registering the node.
result = requests.post(
f'https://127.0.0.1:{daemon_api_port}/register/?serverName={hostname}',
headers={'Authorization': f'Bearer {api_token}'},
verify=False,
timeout=5,
json=result_json,
)
if not result.ok:
return (None, None, f"registration request failed, server returned HTTP {result.status_code}: {result.text}")
return (hostname, api_token, None)
except:
return (None, None, "request failed: " + my_exec_info_message(sys.exc_info()))
def register_callback(self, result: Tuple[str, str, str]):
#print("register_callback", result)
self.register_button.setEnabled(True)
server_name = result[0]
api_token = result[1]
error = result[2]
if error is not None and error is not "":
self.error_popup("Registration Error", "Could not register this computer as a server because:\n\n" + error)
else:
# get_status_callback will display the main screen now that the user is registered (api token is saved)
self.get_status(self.get_status_callback, update_tenant_info=True, must_succeed=True)
def set_config_dirty(self):
if self.apply_config_button is not None:
self.apply_config_button.setDisabled(False)
def apply_config(self):
self.apply_config_button.setDisabled(True)
self.show_apply_config_dialog()
apply_config_worker = Worker(self.apply_config_task)
apply_config_worker.signals.result.connect(self.apply_config_callback)
QThreadPool.globalInstance().start(apply_config_worker)
def apply_config_task(self) -> str:
try:
tunnel_configs = []
for tunnel in self.tunnels:
tunnel_config = dict()
for prop, value in tunnel.items():
if not prop.startswith("gui"):
tunnel_config[prop] = value
tunnel_configs.append(tunnel_config)
result = requests.post(
f'https://127.0.0.1:{daemon_api_port}/apply_config',
verify=False,
timeout=5,
json=tunnel_configs,
)
if not result.ok:
return f"Apply configuration failed, greenhouse background service returned HTTP {result.status_code}: {result.text}"
return None
except:
return "Apply configuration failed: " + my_exec_info_message(sys.exc_info())
def apply_config_callback(self, errorMessage: str):
if errorMessage is not None and errorMessage is not "":
#TODO test this
if self.apply_config_dialog is not None:
self.apply_config_dialog.close()
self.apply_config_dialog = None
self.error_popup("Background Service Error", errorMessage)
def error_popup(self, title: str, error_message: str):
detailedError = None
# truncate the error message if its way too long
if error_message != None and len(error_message) > 250:
detailedError = error_message
error_message = error_message[:250]+" ...truncated, click Show Details to see the rest of the error message"
msg = QMessageBox()
msg.setIcon(QMessageBox.Warning)
msg.setText(error_message)
#msg.setInformativeText("")
msg.setWindowTitle(title)
if detailedError is not None:
msg.setDetailedText(detailedError)
msg.setStandardButtons(QMessageBox.Ok)
msg.exec_()
def create_initial_loading_frame(self):
vbox = QVBoxLayout()
spinner = self.make_waiting_spinner(self, numberOfLines=10, size=5, innerRadius=10)
vbox.addWidget(spinner)
vbox.setAlignment(spinner, Qt.AlignHCenter)
loading_label = QLabel()
loading_label.setText("contacting greenhouse...")
loading_label.setObjectName("loading_label")
vbox.addWidget(loading_label)
vbox.setAlignment(loading_label, Qt.AlignHCenter)
frame_to_return = QFrame()
frame_to_return.setLayout(vbox)
return frame_to_return
def make_waiting_spinner(self, parent, numberOfLines: int, size: int, innerRadius: int):
spinner = WaitingSpinner(parent)
spinner.setRoundness(70.0)
spinner.setMinimumTrailOpacity(15.0)
spinner.setTrailFadePercentage(70.0)
spinner.setNumberOfLines(numberOfLines)
spinner.setLineLength(size*2)
spinner.setLineWidth(size)
spinner.setInnerRadius(innerRadius)
spinner.setRevolutionsPerSecond(0.6)
spinner.setColor(QColor(25,25,25))
spinner.start()
return spinner
def create_login_frame(self, logo: str):
logo_label = QLabel()
logo_label.setPixmap(QPixmap(logo))
header_label = QLabel()
header_label.setText("Register this Computer as a Server")
header_label.setObjectName("header_label")
hostname_label = QLabel()
hostname_label.setText("Server Name")
hostname_input = QLineEdit()
hostname_input.setText(socket.gethostname())
hostname_row = QHBoxLayout()
hostname_layout = QVBoxLayout()
hostname_layout.setObjectName("hostname_layout")
hostname_layout.addWidget(hostname_label)
hostname_layout.addWidget(hostname_input)
hostname_row.addStretch()
hostname_row.addLayout(hostname_layout)
hostname_row.addStretch()
api_token_label = QLabel()
api_token_label.setText("Greenhouse API Token")
api_token_input = PasswordEdit()
# we have to call focus() after the element is displayed for the first time..
# this is a dirty hack to try to make that happen
api_token_focus_wait = Worker(lambda: time.sleep(0.1))
api_token_focus_wait.signals.finished.connect(lambda: api_token_input.setFocus())
QThreadPool.globalInstance().start(api_token_focus_wait)
api_token_row = QHBoxLayout()
api_token_layout = QVBoxLayout()
api_token_layout.setObjectName("api_token_layout")
api_token_layout.addWidget(api_token_label)
api_token_layout.addWidget(api_token_input)
api_token_row.addStretch()
api_token_row.addLayout(api_token_layout)
api_token_row.addStretch()
self.register_button = QPushButton('Register')
self.register_button.clicked.connect(lambda: self.register_clicked(hostname_input.text(), api_token_input.text()))
vbox = QVBoxLayout()
vbox.addStretch()
vbox.addWidget(logo_label)
vbox.setAlignment(logo_label, Qt.AlignHCenter)
vbox.addWidget(header_label)
vbox.setAlignment(header_label, Qt.AlignHCenter)
vbox.addLayout(hostname_row)
vbox.setAlignment(hostname_row, Qt.AlignHCenter)
vbox.addLayout(api_token_row)
vbox.setAlignment(api_token_row, Qt.AlignHCenter)
vbox.addWidget(self.register_button)
vbox.setAlignment(self.register_button, Qt.AlignHCenter)
vbox.addStretch()
frame_to_return = QFrame()
frame_to_return.setLayout(vbox)
return frame_to_return
def create_main_frame(self):
vbox = QVBoxLayout()
if self.status is None:
error_label = QLabel()
error_label.setText("no service status provided")
vbox.addWidget(error_label)
frame_to_return = QFrame()
frame_to_return.setLayout(vbox)
return frame_to_return
underlying_services_status_label = QLabel()
underlying_services_status_label.setText("Underlying Services Status:")
vbox.addWidget(underlying_services_status_label)
self.underlying_services_parent = QVBoxLayout()
self.update_underlying_services_status()
vbox.addLayout(self.underlying_services_parent)
tunnels_label = QLabel()
tunnels_label.setText("Tunnels:")
vbox.addWidget(tunnels_label)
self.gui_tunnels_list = self.create_tunnels_list()
self.tunnels_frame_layout = QVBoxLayout()
self.tunnels_frame_layout.addWidget(self.gui_tunnels_list)
self.tunnels_frame_layout.setAlignment(self.gui_tunnels_list, Qt.AlignTop)
#self.tunnels_frame_layout.addStr
tunnels_frame = QFrame()
tunnels_frame.setLayout(self.tunnels_frame_layout)
scroll_area = QScrollArea()
scroll_area.setVerticalScrollBarPolicy(Qt.ScrollBarAlwaysOn)
scroll_area.setHorizontalScrollBarPolicy(Qt.ScrollBarAlwaysOff)
scroll_area.setWidgetResizable(True)
scroll_area.setWidget(tunnels_frame)
vbox.addWidget(scroll_area, 2)
actions_row = QHBoxLayout()
actions_row.addStretch()
create_tunnel_button = QPushButton('Create Tunnel')
create_tunnel_button.setIcon(QIcon(self.icon_create))
create_tunnel_button.clicked.connect(self.create_tunnel)
actions_row.addWidget(create_tunnel_button)
self.apply_config_button = QPushButton('Apply Configuration Changes')
#create_tunnel_button.setIcon(QIcon(self.icon_create))
self.apply_config_button.clicked.connect(self.apply_config)
actions_row.addWidget(self.apply_config_button)
vbox.addLayout(actions_row)
frame_to_return = QFrame()
frame_to_return.setLayout(vbox)
return frame_to_return
def update_underlying_services_status(self):
if self.underlying_services_frame is not None:
self.underlying_services_parent.removeWidget(self.underlying_services_frame)
sip.delete(self.underlying_services_frame)
self.underlying_services_frame = None
self.underlying_services_frame = QFrame()
hbox = QHBoxLayout()
if self.status is not None:
hbox.addWidget(self.create_service_status("Caddy Server", "Automatic HTTPS Certificates via Let's Encrypt", self.status["caddy"]))
hbox.addWidget(self.create_service_status("Threshold Client", "Greenhouse's Network Tunnel", self.status["threshold"]))
self.underlying_services_frame.setLayout(hbox)
self.underlying_services_parent.addWidget(self.underlying_services_frame)
def create_service_status(self, service_name: str, description: str, service: dict) -> QFrame:
vbox = QVBoxLayout()
service_label = QLabel()
service_label.setText(service_name)
service_label.setObjectName("service_label")
vbox.addWidget(service_label)
hbox = QHBoxLayout()
hbox.addSpacing(10)
inner = QVBoxLayout()
hbox.addLayout(inner)
hbox.addStretch()
vbox.addLayout(hbox)
service_description = QLabel()
service_description.setText(description)
service_description.setObjectName("service_description")
inner.addWidget(service_description)
process_status = QLabel()
health_status = QLabel()
#print(service)
if service["enabled"]:
if service["crash_loop"]:
process_status.setText("Process: unable to start (keeps on crashing)")
elif service["pid"] == 0:
process_status.setText("Process: starting")
else:
process_status.setText(f"Process: running (PID: {service['pid']})")
if service["healthy"]:
health_status.setText("Health Check: healthy")
else:
health_status.setText("Health Check: unhealthy")
else:
process_status.setText("Process: stopped (service is disabled)")
health_status.setText("Health Check: N/A (service is disabled)")
inner.addWidget(process_status)
inner.addWidget(health_status)
logs_button = QPushButton('View Logs...')
logs_button.clicked.connect(lambda: print("display logs"))
inner.addWidget(logs_button)
frame_to_return = QFrame()
frame_to_return.setLayout(vbox)
frame_to_return.setObjectName("rounded_border")
return frame_to_return
def create_tunnels_list(self) -> QFrame:
vbox = QVBoxLayout()
for tunnel in self.tunnels:
#print("tunnel:", tunnel)
row = self.create_tunnel_row(tunnel)
vbox.addWidget(row)
vbox.addStretch()
frame_to_return = QFrame()
frame_to_return.setLayout(vbox)
return frame_to_return
def create_tunnel_row(self, tunnel: dict) -> QFrame:
authorized_domains = self.status["tenant_info"]["AuthorizedDomains"]
port_start = self.status["tenant_info"]["PortStart"]
port_end = self.status["tenant_info"]["PortEnd"]
disabled_palette = QPalette()
disabled_palette.setColor(QPalette.Text, Qt.gray)
# there is no border color on QPalette 😢 https://doc.qt.io/qt-5/qpalette.html
enabled_palette = QPalette()
enabled_palette.setColor(QPalette.Text, Qt.black)
# the protocol selector will be built after some other stuff that it depends on is set up,
# see protocol_dropdown below.
any_domain_column = QVBoxLayout()
any_domain_label = QLabel()
any_domain_label.setText("domain (not editable because TCP routes by port)")
any_domain_input = QLineEdit()
any_domain_input.setText(authorized_domains[len(authorized_domains)-1])
any_domain_input.setReadOnly(True)
any_domain_input.setObjectName("any_domain_field")
any_domain_column.addWidget(any_domain_label)
any_domain_column.addWidget(any_domain_input)
tunnel['gui_any_domain_frame'] = QFrame()
tunnel['gui_any_domain_frame'].setLayout(any_domain_column)
domain_column = QVBoxLayout()
labels_row = QHBoxLayout()
subdomain_label = QLabel()
subdomain_label.setText("subdomain")
subdomain_checkbox = QCheckBox()
subdomain_checkbox.setChecked('has_subdomain' in tunnel and tunnel['has_subdomain'])
def set_has_subdomain_on_tunnel(tunnel: dict, has_subdomain: bool):
tunnel['has_subdomain'] = has_subdomain
self.set_config_dirty()
if has_subdomain:
tunnel['gui_subdomain_input'].show()
tunnel['gui_dot_label'].show()
else:
tunnel['gui_subdomain_input'].hide()
tunnel['gui_dot_label'].hide()
subdomain_checkbox.stateChanged.connect(lambda i: set_has_subdomain_on_tunnel(tunnel, subdomain_checkbox.isChecked()))
labels_row.addWidget(subdomain_label)
labels_row.addWidget(subdomain_checkbox)
labels_row.addStretch()
domain_label = QLabel()
domain_label.setText("domain")
labels_row.addWidget(domain_label)
domain_column.addLayout(labels_row)
fields_row = QHBoxLayout()
def set_subdomain_on_tunnel(tunnel: dict, subdomain: str):
tunnel['subdomain'] = subdomain
self.set_config_dirty()
tunnel['gui_subdomain_input'] = QLineEdit()
tunnel['gui_subdomain_input'].setObjectName("subdomain_input")
tunnel['gui_subdomain_input'].setText(tunnel['subdomain'] if 'subdomain' in tunnel else "")
tunnel['gui_subdomain_input'].textChanged.connect(lambda i: set_subdomain_on_tunnel(tunnel, tunnel['gui_subdomain_input'].text()))
tunnel['gui_dot_label'] = QLabel()
tunnel['gui_dot_label'].setText(".")
def set_domain_on_tunnel(tunnel: dict, domain: str):
tunnel['domain'] = domain
self.set_config_dirty()
domain_dropdown = QComboBox()
domain_dropdown.setObjectName("domain_dropdown")
domain_dropdown.currentIndexChanged.connect(lambda i: set_domain_on_tunnel(tunnel, authorized_domains[i]))
for authorized_domain in authorized_domains:
domain_dropdown.addItem(authorized_domain)
fields_row.addWidget(tunnel['gui_subdomain_input'])
fields_row.addWidget(tunnel['gui_dot_label'])
fields_row.addWidget(domain_dropdown)
domain_column.addLayout(fields_row)
tunnel['gui_domain_frame'] = QFrame()
tunnel['gui_domain_frame'].setLayout(domain_column)
has_subdomain = (True if 'has_subdomain' in tunnel and tunnel['has_subdomain'] else False)
subdomain_checkbox.setChecked(has_subdomain)
set_has_subdomain_on_tunnel(tunnel, has_subdomain)
public_port_column = QVBoxLayout()
public_port_label = QLabel()
public_port_label.setText(" port")
def set_public_arbitrary_port_on_tunnel(tunnel: dict, port_str: str):
current_public_port = 443
if "public_port" in tunnel:
current_public_port = int(tunnel['public_port'])
if port_str.isdigit():
tunnel['public_port'] = int(port_str)
self.set_config_dirty()
tunnel['gui_public_arbitrary_port_input'].setText(str(int(port_str)))
else:
tunnel['gui_public_arbitrary_port_input'].setText(str(current_public_port))
tunnel['gui_public_arbitrary_port_input'] = QLineEdit()
tunnel['gui_public_arbitrary_port_input'].setObjectName("port_field")
tunnel['gui_public_arbitrary_port_input'].setText("443")
tunnel['gui_public_arbitrary_port_input'].textEdited.connect(lambda i: set_public_arbitrary_port_on_tunnel(tunnel, tunnel['gui_public_arbitrary_port_input'].text()))
def set_public_port_on_tunnel(tunnel: dict, port: int):
tunnel['public_port'] = port
self.set_config_dirty()
tunnel['gui_public_port_dropdown'] = QComboBox()
for port in range(port_start, port_end):
tunnel['gui_public_port_dropdown'].addItem(str(port))
domain_dropdown.currentIndexChanged.connect(lambda i: set_public_port_on_tunnel(tunnel, port_start+i))
public_port_colon_label = QLabel()
public_port_colon_label.setText(":")
public_port_row = QHBoxLayout()
public_port_row.addWidget(public_port_colon_label)
public_port_row.setAlignment(public_port_colon_label, Qt.AlignLeft)
public_port_row.addWidget(tunnel['gui_public_arbitrary_port_input'])
public_port_row.addWidget(tunnel['gui_public_port_dropdown'])
public_port_column.addWidget(public_port_label)
public_port_column.addLayout(public_port_row)
public_port_column.setAlignment(public_port_row, Qt.AlignLeft)
tunnel['gui_public_port_frame'] = QFrame()
tunnel['gui_public_port_frame'].setLayout(public_port_column)
# end of the source row, beginning of the protocol selector and destination row stuff
# the protocol selector determines the availiable options for the destination type dropdown,
# so the function that builds the destination type dropdown has to be defined before we define the protocol selector.
def recreate_destination_type_dropdown(tunnel: dict, allow_folder: bool):
if 'gui_type_dropdown' in tunnel and tunnel['gui_type_dropdown'] is not None:
try:
tunnel['gui_type_column'].removeWidget(tunnel['gui_type_dropdown'])
sip.delete(tunnel['gui_type_dropdown'])
tunnel['gui_type_dropdown'] = None
except:
pass
destination_types = [
dict(key="local_port", description="forward to local listener", label=""),
dict(key="host_port", description="forward to another server (LAN)"),
]
if allow_folder:
destination_types.append(dict(key="folder", description="serve local files"))
tunnel['gui_type_dropdown'] = QComboBox()
for destination_type in destination_types:
tunnel['gui_type_dropdown'].addItem(destination_type["description"])
def set_destination_type_on_tunnel(tunnel: dict, i: int):
destination_type = destination_types[i]['key']
tunnel['destination_type'] = destination_type
self.set_config_dirty()
if destination_type == "local_port" or destination_type == "host_port":
tunnel['gui_dest_folder_frame'].hide()
tunnel['gui_dest_socket_frame'].show()
if destination_type == "local_port":
tunnel['gui_choose_listener_button'].show()
tunnel['gui_host_field'].setText("localhost")
tunnel['gui_host_field'].setReadOnly(True)
tunnel['gui_host_field'].setPalette(disabled_palette)
else:
tunnel['gui_choose_listener_button'].hide()
tunnel['gui_host_field'].setReadOnly(False)
tunnel['gui_host_field'].setPalette(enabled_palette)
else:
tunnel['gui_dest_folder_frame'].show()
tunnel['gui_dest_socket_frame'].hide()
tunnel['gui_type_dropdown'].currentIndexChanged.connect(lambda i: set_destination_type_on_tunnel(tunnel, i))
def initialize_destination_type_for_tunnel(tunnel):
destination_type_was_set = False
for i in range(len(destination_types)):
if 'destination_type' in tunnel and destination_types[i]['key'] == tunnel['destination_type']:
tunnel['gui_type_dropdown'].setCurrentIndex(i)
set_destination_type_on_tunnel(tunnel, i)
destination_type_was_set = True
if destination_type_was_set == False:
set_destination_type_on_tunnel(tunnel, 0)
# we have to call setPalette() after the element is displayed for the first time..
# this is a dirty hack to try to make that happen
initialize_destination_type_wait = Worker(lambda: time.sleep(0.1))
initialize_destination_type_wait.signals.finished.connect(lambda: initialize_destination_type_for_tunnel(tunnel))
QThreadPool.globalInstance().start(initialize_destination_type_wait)
# its idempotent so we can also call it right away to prevent the 0.1 second visual glitch when adding a new tunnel
initialize_destination_type_for_tunnel(tunnel)
tunnel['gui_type_column'].addWidget(tunnel['gui_type_dropdown'])
## end of the destination type dropdown, start of the tunnel protocol selector
protocol_column = QVBoxLayout()
protocol_label = QLabel()
protocol_label.setText("protocol")
protocol_column.addWidget(protocol_label)
protocols = [
dict(description="HTTPS", key="https"),
dict(description="TLS (any)", key="tls"),
dict(description="TCP (any)", key="tcp"),
]
protocol_dropdown = QComboBox()
for protocol in protocols:
protocol_dropdown.addItem(protocol["description"])
def set_protocol_on_tunnel(tunnel: dict, protocol: str):
previous_allow_folder = 'protocol' not in tunnel or tunnel['protocol'] == "https"
new_allow_folder = protocol == "https"
if previous_allow_folder != new_allow_folder:
recreate_destination_type_dropdown(tunnel, new_allow_folder)
tunnel['protocol'] = protocol
self.set_config_dirty()
#print(protocol)
if protocol == "https":
tunnel['public_port'] = 443
tunnel['gui_public_arbitrary_port_input'].setText("443")
tunnel['gui_public_port_frame'].hide()
else:
tunnel['gui_public_port_frame'].show()
if protocol == "tcp":
# TODO add a checkbox for PROXY protocol when backend is TCP
tunnel['gui_any_domain_frame'].show()
tunnel['gui_domain_frame'].hide()
tunnel['gui_public_arbitrary_port_input'].hide()
tunnel['gui_public_port_dropdown'].show()
else:
tunnel['gui_any_domain_frame'].hide()
tunnel['gui_domain_frame'].show()
tunnel['gui_public_arbitrary_port_input'].show()
tunnel['gui_public_port_dropdown'].hide()
protocol_was_set = False
for i in range(len(protocols)):
if 'protocol' in tunnel and protocols[i]['key'] == tunnel['protocol']:
protocol_dropdown.setCurrentIndex(i)
set_protocol_on_tunnel(tunnel, protocols[i]['key'])
protocol_was_set = True
if protocol_was_set == False:
set_protocol_on_tunnel(tunnel, protocols[0]['key'])
protocol_dropdown.currentIndexChanged.connect(lambda i: set_protocol_on_tunnel(tunnel, protocols[i]['key']))
protocol_row = QHBoxLayout()
colon_slash_slash_label = QLabel()
colon_slash_slash_label.setText("://")
protocol_row.addWidget(protocol_dropdown)
protocol_row.addWidget(colon_slash_slash_label)
protocol_column.addLayout(protocol_row)
source_row = QHBoxLayout()
source_row.addStretch()
source_row.addLayout(protocol_column)
source_row.setAlignment(protocol_column, Qt.AlignVCenter)
source_row.addWidget(tunnel['gui_any_domain_frame'])
source_row.setAlignment(tunnel['gui_any_domain_frame'], Qt.AlignVCenter)
source_row.addWidget(tunnel['gui_domain_frame'])
source_row.setAlignment(tunnel['gui_domain_frame'], Qt.AlignVCenter)
source_row.addWidget(tunnel['gui_public_port_frame'])
source_row.setAlignment(tunnel['gui_public_port_frame'], Qt.AlignVCenter )
row = QHBoxLayout()
source_and_sink = QVBoxLayout()
source_and_sink.addLayout(source_row)
source_and_sink.setAlignment(source_row, Qt.AlignTop)
sink_row = QHBoxLayout()
sink_row.addStretch()
# the destination type selector will be added after we build up the destination gui elements
dest_tunnel_row = QHBoxLayout()
host_column = QVBoxLayout()
host_label = QLabel()
host_label.setText("hostname")
def set_destination_hostname_on_tunnel(tunnel: dict, hostname: str):
tunnel['destination_hostname'] = hostname
self.set_config_dirty()
tunnel['gui_host_field'] = QLineEdit()
tunnel['gui_host_field'].setObjectName("host_field")
tunnel['gui_host_field'].textEdited.connect(lambda: set_destination_hostname_on_tunnel(tunnel, tunnel['gui_host_field'].text()))
colon_label = QLabel()
colon_label.setText(":")
host_row = QHBoxLayout()
host_row.addWidget(tunnel['gui_host_field'])
host_row.addWidget(colon_label)
host_column.addWidget(host_label)
host_column.addLayout(host_row)
port_column = QVBoxLayout()
port_label = QLabel()
port_label.setText("port")
def set_destination_port_on_tunnel(tunnel: dict, port_str: str):
current_destination_port = 80
if "destination_port" in tunnel:
current_destination_port = tunnel['destination_port']
if port_str.isdigit():
tunnel['destination_port'] = int(port_str)
self.set_config_dirty()
tunnel['gui_destination_port_field'].setText(str(int(port_str)))
else:
tunnel['gui_destination_port_field'].setText(str(current_destination_port))
if "destination_port" not in tunnel:
tunnel['destination_port'] = 80
tunnel['gui_destination_port_field'] = QLineEdit()
tunnel['gui_destination_port_field'].setObjectName("port_field")
tunnel['gui_destination_port_field'].setText(str(tunnel['destination_port']))
tunnel['gui_destination_port_field'].textEdited.connect(lambda: set_destination_port_on_tunnel(tunnel, tunnel['gui_destination_port_field'].text()))
tunnel['gui_choose_listener_button'] = QPushButton('Choose...')
tunnel['gui_choose_listener_button'].setIcon(QIcon(self.icon_open_socket))
def choose_listener_for_tunnel(tunnel: dict):
print("choose_listener_for_tunnel")
tunnel['gui_choose_listener_button'].clicked.connect(lambda: choose_listener_for_tunnel(tunnel))
port_row = QHBoxLayout()
port_row.addWidget(tunnel['gui_destination_port_field'])
port_row.addWidget(tunnel['gui_choose_listener_button'])
port_column.addWidget(port_label)
port_column.addLayout(port_row)
dest_tunnel_row.addLayout(host_column)
dest_tunnel_row.addLayout(port_column)
tunnel['gui_dest_socket_frame'] = QFrame()
tunnel['gui_dest_socket_frame'].setLayout(dest_tunnel_row)
dest_folder_column = QVBoxLayout()
folder_label = QLabel()
folder_label.setText("folder path")
def set_folder_path_on_tunnel(tunnel: dict, folder_path: str):
tunnel['folder_path'] = folder_path
self.set_config_dirty()
folder_field = QLineEdit()
folder_field.setText("")
folder_field.setObjectName("folder_field")
folder_field.textEdited.connect(lambda: set_folder_path_on_tunnel(tunnel, folder_field.text()))
choose_folder_button = QPushButton('Choose...')
choose_folder_button.setIcon(QIcon(self.icon_open_folder))
def choose_listener_for_tunnel(tunnel: dict):
print("choose_folder_for_tunnel")
choose_folder_button.clicked.connect(lambda: choose_listener_for_tunnel(tunnel))
folder_row = QHBoxLayout()
folder_row.addWidget(folder_field)
folder_row.addWidget(choose_folder_button)
dest_folder_column.addWidget(folder_label)
dest_folder_column.addLayout(folder_row)
tunnel['gui_dest_folder_frame'] = QFrame()
tunnel['gui_dest_folder_frame'].setLayout(dest_folder_column)
# end of destination row, now we can build the destination type selector.
tunnel['gui_type_column'] = QVBoxLayout()
type_label = QLabel()
type_label.setText("type")
tunnel['gui_type_column'].addWidget(type_label)
# see recreate_destination_type_dropdown above
allow_folder_destination_type = 'protocol' not in tunnel or tunnel['protocol'] == "https"
recreate_destination_type_dropdown(tunnel, allow_folder_destination_type)
sink_row.addLayout(tunnel['gui_type_column'])
sink_row.setAlignment(tunnel['gui_type_column'], Qt.AlignVCenter)
sink_row.addWidget(tunnel['gui_dest_socket_frame'])
sink_row.setAlignment(tunnel['gui_dest_socket_frame'], Qt.AlignVCenter)
sink_row.addWidget(tunnel['gui_dest_folder_frame'])
sink_row.setAlignment(tunnel['gui_dest_folder_frame'], Qt.AlignVCenter)
source_and_sink.addLayout(sink_row)
source_and_sink.setAlignment(sink_row, Qt.AlignTop)
#source_and_sink.addStretch()
row.addLayout(source_and_sink)
arrow_column = QVBoxLayout()
arrow_image_label = QLabel()
arrow_image_label.setPixmap(QPixmap(self.image_tunnel_arrow))
arrow_column.addStretch()
arrow_column.addWidget(arrow_image_label)
arrow_column.addSpacing(4)
row.addLayout(arrow_column)
row.addStretch()
to_return = QFrame()
outer_row = QHBoxLayout()
tunnel_frame = QFrame()
tunnel_frame.setObjectName("rounded_border")
tunnel_frame.setLayout(row)
outer_row.addWidget(tunnel_frame)
outer_row.addStretch()
to_return.setLayout(outer_row)
return to_return
def show_apply_config_dialog(self):
self.apply_config_button.setDisabled(True)
self.apply_config_dialog = QDialog()
self.apply_config_dialog_layout = QVBoxLayout()
self.apply_config_dialog_content = QFrame()
loading_layout = QVBoxLayout()
spinner = self.make_waiting_spinner(None, numberOfLines=10, size=5, innerRadius=10)
loading_layout.addWidget(spinner)
loading_layout.setAlignment(spinner, Qt.AlignHCenter)
loading_label = QLabel()
loading_label.setText("contacting background services...")
loading_label.setObjectName("loading_label")
loading_layout.addWidget(loading_label)
loading_layout.setAlignment(loading_label, Qt.AlignHCenter)
self.apply_config_dialog_content.setLayout(loading_layout)
self.apply_config_dialog_layout.addWidget(self.apply_config_dialog_content)
self.apply_config_dialog.setLayout(self.apply_config_dialog_layout)
dialog_rect = QRect(0, 0, 300, 500)
window_rect = self.frameGeometry()
dialog_rect.moveCenter(window_rect.center())
self.apply_config_dialog.setGeometry(dialog_rect.left(), dialog_rect.top(), dialog_rect.width(), dialog_rect.height())
self.apply_config_dialog.show()
def update_apply_config_dialog(self):
current_error = ""
if "apply_config_status_error" in self.status:
current_error = self.status["apply_config_status_error"]
different_index = self.last_apply_config_status_index != self.status["apply_config_status_index"]
different_error = self.last_apply_config_status_error != current_error
if not different_index and not different_error:
return
if self.apply_config_dialog_content is not None:
self.apply_config_dialog_layout.removeWidget(self.apply_config_dialog_content)
sip.delete(self.apply_config_dialog_content)
self.apply_config_dialog_content = None
self.apply_config_dialog_content = QFrame()
vbox = QVBoxLayout()
vbox.addStretch()
i = 0
for apply_config_status in self.status["apply_config_statuses"]:
hbox = QHBoxLayout()
if self.status["apply_config_status_index"] > i:
checkmark_label = QLabel()
checkmark_label.setPixmap(QPixmap(self.icon_success))
checkmark_label.setScaledContents(True)
one_em_size = checkmark_label.fontMetrics().height()
#checkmark_label.setSizePolicy(QSizePolicy(QSizePolicy.Ignored, QSizePolicy.Ignored))
checkmark_label.setFixedSize(one_em_size, one_em_size)
checkmark_label.setObjectName("icon_label")
hbox.addWidget(checkmark_label)
hbox.setAlignment(checkmark_label, Qt.AlignVCenter)
elif self.status["apply_config_status_index"] == i:
if "apply_config_status_error" in self.status and self.status["apply_config_status_error"] != "":
failure_label = QLabel()
failure_label.setPixmap(QPixmap(self.icon_failure))
failure_label.setScaledContents(True)
one_em_size = failure_label.fontMetrics().height()
#checkmark_label.setSizePolicy(QSizePolicy(QSizePolicy.Ignored, QSizePolicy.Ignored))
failure_label.setFixedSize(one_em_size, one_em_size)
failure_label.setObjectName("icon_label")
hbox.addWidget(failure_label)
hbox.setAlignment(failure_label, Qt.AlignVCenter)
else:
# TODO: figure out how to make spinner sit in the right place.
# spinner = self.make_waiting_spinner(None, numberOfLines=5, size=2, innerRadius=3)
# hbox.addWidget(spinner)
# hbox.setAlignment(spinner, Qt.AlignVCenter)
hbox.addSpacing(10)
else:
hbox.addSpacing(10)
task_label = QLabel()
task_label.setText(apply_config_status)
task_label.setObjectName("apply_config_status_label")
hbox.addWidget(task_label)
hbox.setAlignment(task_label, Qt.AlignVCenter)
vbox.addLayout(hbox)
i += 1
vbox.addStretch()
self.apply_config_dialog_content.setLayout(vbox)
self.apply_config_dialog_layout.addWidget(self.apply_config_dialog_content)
def handle_close_window():
if self.apply_config_dialog is not None:
self.apply_config_dialog.close()
self.apply_config_dialog = None
if self.status["apply_config_status_index"] >= len(self.status["apply_config_statuses"]):
close_window_wait = Worker(lambda: time.sleep(1))
close_window_wait.signals.finished.connect(handle_close_window)
QThreadPool.globalInstance().start(close_window_wait)
if __name__ == '__main__':
app_context = ApplicationContext()
stylesheet = app_context.get_resource('styles.qss')
images = dict(
logo=app_context.get_resource('greenhouse.png'),
icon_create = app_context.get_resource('icon_create.png'),
icon_open_socket = app_context.get_resource('icon_open_socket.png'),
icon_open_folder = app_context.get_resource('icon_open_folder.png'),
icon_success = app_context.get_resource('icon_success.png'),
icon_failure = app_context.get_resource('icon_failure.png'),
image_tunnel_arrow = app_context.get_resource('tunnel_arrow.png'),
)
app_context.app.setStyleSheet(open(stylesheet).read())
window = MainWindow(images)
window.show()
window.resize(900, 700)
exit_code = app_context.app.exec_()
sys.exit(exit_code)