diff --git a/linux_gui/presenters/dashboard_presenter.py b/linux_gui/presenters/dashboard_presenter.py index 6dfc1b8..980df67 100644 --- a/linux_gui/presenters/dashboard_presenter.py +++ b/linux_gui/presenters/dashboard_presenter.py @@ -1,38 +1,32 @@ -import time import datetime -import requests -import subprocess -import collections -import concurrent.futures - -# Remote imports -from protonvpn_cli.utils import ( - get_config_value, - is_connected, - get_servers, - get_server_value, - get_transferred_data, - pull_server_data, - get_country_name -) -from protonvpn_cli.country_codes import country_codes #noqa +import time # PyGObject import import gi gi.require_version('Gtk', '3.0') -from gi.repository import GObject as gobject, GdkPixbuf +from gi.repository import GObject as gobject, GdkPixbuf # noqa + +from protonvpn_cli.country_codes import country_codes # noqa + +# Remote imports +from protonvpn_cli.utils import ( # noqa + get_config_value, + is_connected, + get_servers, + get_server_value, + get_transferred_data, + pull_server_data +) # Local imports -from ..gui_logger import gui_logger -from ..constants import GITHUB_URL_RELEASE, VERSION, LARGE_FLAGS_BASE_PATH, SMALL_FLAGS_BASE_PATH, FEATURES_BASE_PATH -from ..utils import ( - get_server_protocol_from_cli, +from ..gui_logger import gui_logger # noqa +from ..constants import VERSION, LARGE_FLAGS_BASE_PATH # noqa +from ..utils import ( # noqa get_gui_config, - set_gui_config, - check_internet_conn, custom_get_ip_info, ) + class DashboardPresenter: def __init__(self, dashboard_service, queue): self.dashboard_service = dashboard_service @@ -44,7 +38,7 @@ def on_load(self, objects_dict): display_message = "Could not load necessary resources, there might be connectivity issues." time.sleep(2) self.queue.put(dict(action="hide_spinner")) - + conn = custom_get_ip_info() if conn and not conn is None: @@ -57,11 +51,12 @@ def on_load_dashboard_content(self, objects_dict, conn): self.queue.put(dict(action="hide_dialog")) # Sets Secure-Core switch state - self.on_load_set_secure_core(objects_dict["secure_core"]["secure_core_switch"], objects_dict["secure_core"]["secure_core_label_style"]) + self.on_load_set_secure_core(objects_dict["secure_core"]["secure_core_switch"], + objects_dict["secure_core"]["secure_core_label_style"]) # Loads the labels self.on_update_labels(objects_dict["connection_labels"]) - + # Loads the server list self.on_update_server_list(objects_dict["server_tree_list"]["tree_object"]) @@ -73,7 +68,7 @@ def on_load_set_secure_core(self, secure_core_switch, secure_core_label): except KeyError: display_secure_core = "False" - secure_core_label_style = secure_core_label.get_style_context() + secure_core_label_style = secure_core_label.get_style_context() if display_secure_core == "True": secure_core_switch.set_state(True) @@ -83,7 +78,7 @@ def on_load_set_secure_core(self, secure_core_switch, secure_core_label): def reload_secure_core_servers(self, **kwargs): """Function that reloads server list to either secure-core or non-secure-core. - """ + """ # Sleep is needed because it takes a second to update the information, # which makes the button "lag". Temporary solution. time.sleep(1) @@ -94,13 +89,14 @@ def reload_secure_core_servers(self, **kwargs): return_val = True self.on_update_server_list(kwargs.get("tree_object")) - display_message = "Displaying {} servers!".format("secure-core" if kwargs.get("update_to") == "True" else "non secure-core") + display_message = "Displaying {} servers!".format( + "secure-core" if kwargs.get("update_to") == "True" else "non secure-core") self.queue.put(dict(action="update_dialog", label=display_message)) def on_connect_user_selected(self, **kwargs): """Function that either connects by selected server or selected country. - """ + """ user_selected_server = kwargs.get("user_selected_server") # Check if it should connect to country or server @@ -115,20 +111,20 @@ def on_connect_user_selected(self, **kwargs): def quick_connect(self, **kwargs): """Function that connects to the quickest server. - """ + """ profile_quick_connect = False if "profile_quick_connect" in kwargs: profile_quick_connect = True display_message = self.dashboard_service.quick_connect_manager(profile_quick_connect) - + self.queue.put(dict(action="update_dialog", label=display_message)) - + self.on_update_labels(kwargs.get("connection_labels")) def on_last_connect(self, **kwargs): """Function that connects to the last connected server. - """ + """ display_message = self.dashboard_service.last_connect() self.queue.put(dict(action="update_dialog", label=display_message)) @@ -146,7 +142,7 @@ def random_connect(self, **kwargs): def on_refresh_servers(self, **kwargs): """Function that reloads server list to either secure-core or non-secure-core. - """ + """ # Sleep is needed because it takes a second to update the information, # which makes the button "lag". Temporary solution. time.sleep(1) @@ -161,7 +157,7 @@ def on_refresh_servers(self, **kwargs): return_val = True else: self.queue.put(dict(action="update_dialog", label="Could not update servers!", spinner=False)) - + def on_disconnect(self, **kwargs): """Function that disconnects from the VPN. """ @@ -177,13 +173,13 @@ def on_check_for_updates(self): """ return_string = "Developer Mode." return_val = False - + try: latest_release, pip3_installed = self.dashboard_service.check_for_updates() except: latest_release = False - if not latest_release: + if not latest_release: return_string = "Failed to check for updates." else: if latest_release == VERSION: @@ -191,7 +187,8 @@ def on_check_for_updates(self): return_val = True if VERSION < latest_release: - return_string = "There is a newer release, you should upgrade to v{0}.\n\n".format(latest_release) + return_string = "There is a newer release, you should upgrade to v{0}.\n\n".format( + latest_release) if pip3_installed: return_string = return_string + "You can upgrade with the following command:\n\nsudo pip3 install protonvpn-linux-gui-calexandru2018 --upgrade\n\n" else: @@ -205,7 +202,7 @@ def on_check_for_updates(self): def on_diagnose(self): """Multipurpose message dialog function. """ - reccomendation, has_internet, is_custom_resolv_conf, is_killswitch_enabled, is_ovpnprocess_running, is_dns_protection_enabled,is_splitunn_enabled = self.dashboard_service.diagnose() + recomendation, has_internet, is_custom_resolv_conf, is_killswitch_enabled, is_ovpnprocess_running, is_dns_protection_enabled, is_splitunn_enabled = self.dashboard_service.diagnose() result = """ Has internet:\t\t\t\t{has_internet} @@ -215,16 +212,17 @@ def on_diagnose(self): DNS Protection Enabled:\t\t{is_dns_enabled} Split Tunneling Enabled:\t\t{is_sp_enabled} """.format( - has_internet= "Yes" if has_internet else "No", + has_internet="Yes" if has_internet else "No", resolv_conf_status=is_custom_resolv_conf["display"], - is_ks_enabled= "Yes" if is_killswitch_enabled else "No", - is_vpnprocess_running= "Yes" if is_ovpnprocess_running else "No", - is_dns_enabled= "Yes" if is_dns_protection_enabled else "No", - is_sp_enabled= "Yes" if is_splitunn_enabled else "No") + is_ks_enabled="Yes" if is_killswitch_enabled else "No", + is_vpnprocess_running="Yes" if is_ovpnprocess_running else "No", + is_dns_enabled="Yes" if is_dns_protection_enabled else "No", + is_sp_enabled="Yes" if is_splitunn_enabled else "No") gui_logger.debug(result) - self.queue.put(dict(action="update_dialog", label=result, sub_label="Reccomendation:\n{recc}".format(recc=reccomendation))) + self.queue.put(dict(action="update_dialog", label=result, + sub_label="Recomendation:\n{recc}".format(recc=recomendation))) def on_update_labels(self, connection_labels_dict, disconnect=False): # Update labels @@ -238,25 +236,25 @@ def update_labels_status(self, connection_labels_dict, disconnecting): """Function prepares data to update labels. """ gui_logger.debug(">>> Running \"update_labels_status\" getting servers, is_connected and connected_server.") - + servers = get_servers() is_vpn_connected = True if is_connected() else False country_cc = False load = False - time_connected_label = connection_labels_dict[0]["time_connected_label"] - protocol_label = connection_labels_dict[0]["protocol_label"] - conn_disc_button_label = connection_labels_dict[0]["conn_disc_button_label"] - ip_label = connection_labels_dict[0]["ip_label"] - server_load_label = connection_labels_dict[0]["server_load_label"] - country_label = connection_labels_dict[0]["country_label"] - isp_label = connection_labels_dict[0]["isp_label"] - data_received_label = connection_labels_dict[0]["data_received_label"] - data_sent_label = connection_labels_dict[0]["data_sent_label"] - background_large_flag = connection_labels_dict[0]["background_large_flag"] - protonvpn_sign_green = connection_labels_dict[0]["protonvpn_sign_green"] - + time_connected_label = connection_labels_dict[0]["time_connected_label"] + protocol_label = connection_labels_dict[0]["protocol_label"] + conn_disc_button_label = connection_labels_dict[0]["conn_disc_button_label"] + ip_label = connection_labels_dict[0]["ip_label"] + server_load_label = connection_labels_dict[0]["server_load_label"] + country_label = connection_labels_dict[0]["country_label"] + isp_label = connection_labels_dict[0]["isp_label"] + data_received_label = connection_labels_dict[0]["data_received_label"] + data_sent_label = connection_labels_dict[0]["data_sent_label"] + background_large_flag = connection_labels_dict[0]["background_large_flag"] + protonvpn_sign_green = connection_labels_dict[0]["protonvpn_sign_green"] + try: connected_server = get_config_value("metadata", "connected_server") except (KeyError, IndexError): @@ -267,7 +265,7 @@ def update_labels_status(self, connection_labels_dict, disconnecting): load = get_server_value(connected_server, "Load", servers) except (KeyError, IndexError): gui_logger.debug("[!] Could not find \"server load\" information.") - + load = "{0}% Load".format(load) if load and is_vpn_connected else "" server_load_label.set_markup('{0}'.format(load)) @@ -277,17 +275,17 @@ def update_labels_status(self, connection_labels_dict, disconnecting): ip, isp, country = result else: ip = "None" - isp = "None" + isp = "None" country = "None" country = country.lower() - for k,v in country_codes.items(): + for k, v in country_codes.items(): if (k.lower() == country) or (k.lower() == "uk" and country == "gb"): if is_vpn_connected: if k.lower() == "uk" and country == "gb": k = "gb" - flag_path = LARGE_FLAGS_BASE_PATH+"{}.jpg".format(k.lower()) + flag_path = LARGE_FLAGS_BASE_PATH + "{}.jpg".format(k.lower()) background_large_flag.set_from_file(flag_path) country_cc = v break @@ -298,7 +296,7 @@ def update_labels_status(self, connection_labels_dict, disconnecting): if is_vpn_connected: try: country_server = country_server + " >> " + connected_server - except TypeError: + except TypeError: country_server = country_server + " >> " protonvpn_sign_green.show() @@ -310,8 +308,10 @@ def update_labels_status(self, connection_labels_dict, disconnecting): country_label.set_markup(country_server if country_server else "") # Update sent and received data - gobject.timeout_add_seconds(1, self.update_sent_received_data, {"is_vpn_connected": is_vpn_connected, "received_label": data_received_label, "sent_label": data_sent_label}) - + gobject.timeout_add_seconds(1, self.update_sent_received_data, + {"is_vpn_connected": is_vpn_connected, "received_label": data_received_label, + "sent_label": data_sent_label}) + # Check and set VPN status label. Get also protocol status if vpn is connected protocol = "No VPN Connection" conn_disc_button = "Quick Connect" @@ -328,7 +328,8 @@ def update_labels_status(self, connection_labels_dict, disconnecting): dns_enabled = get_config_value("USER", "dns_leak_protection") # Update time connected label - gobject.timeout_add_seconds(1, self.update_connection_time, {"is_vpn_connected":is_vpn_connected, "label":time_connected_label}) + gobject.timeout_add_seconds(1, self.update_connection_time, + {"is_vpn_connected": is_vpn_connected, "label": time_connected_label}) # Check and set protocol label protocol_label.set_markup(protocol) @@ -337,18 +338,18 @@ def update_sent_received_data(self, dict_labels): tx_amount, rx_amount = get_transferred_data() rx_amount = rx_amount if dict_labels["is_vpn_connected"] else "" - + dict_labels["received_label"].set_markup('{0}'.format(rx_amount)) # Get and set sent data tx_amount = tx_amount if dict_labels["is_vpn_connected"] else "" dict_labels["sent_label"].set_markup('{0}'.format(tx_amount)) - + return True def update_connection_time(self, dict_data): connection_time = False - + if dict_data["is_vpn_connected"]: try: connected_time = get_config_value("metadata", "connected_time") @@ -356,12 +357,29 @@ def update_connection_time(self, dict_data): connection_time = str(datetime.timedelta(seconds=connection_time)).split(".")[0] except (KeyError, IndexError): connection_time = False - + connection_time = connection_time if connection_time else "" dict_data["label"].set_markup('{0}'.format(connection_time)) return True + def create_country_flag(self, country): + """ + Creates a new GdkPixbuf.Pixbuf object from a file. If img file for specified country is not found (even though + country is in protonvpn_cli.country_codes), it uses Unknow.png file. If this file gets lost somehow, it just + creates white GdkPixbuf.Pixbuf object. + """ + try: + flag = GdkPixbuf.Pixbuf.new_from_file_at_size(self.dashboard_service.get_flag_path(country), 15, 15) + except gi.repository.GLib.Error: + try: + flag = GdkPixbuf.Pixbuf.new_from_file_at_size(self.dashboard_service.get_flag_path(None), 15, 15) + except gi.repository.GLib.Error: + flag = GdkPixbuf.Pixbuf.new(GdkPixbuf.Colorspace.RGB, True, 8, 15, 15) + flag.fill(0xffffffff) + + return flag + def populate_server_list(self, tree_object): """Function that updates server list. """ @@ -379,10 +397,11 @@ def populate_server_list(self, tree_object): for country in country_servers: # Get average load and highest feature - avrg_load, country_feature = self.dashboard_service.get_country_avrg_features(country, country_servers, servers, only_secure_core) + avrg_load, country_feature = self.dashboard_service.get_country_avrg_features(country, country_servers, + servers, only_secure_core) + + flag = self.create_country_flag(country) - flag = GdkPixbuf.Pixbuf.new_from_file_at_size(self.dashboard_service.get_flag_path(country), 15,15) - # Check plus servers if country_feature == "normal" or country_feature == "p2p": plus_feature = images_dict["empty_pix"] @@ -403,11 +422,12 @@ def populate_server_list(self, tree_object): country_row = tree_object.append(None, [flag, country, plus_feature, feature, avrg_load]) for servername in country_servers[country]: - servername, plus_feature, feature, load, secure_core = self.dashboard_service.set_individual_server(servername, images_dict, servers, feature) + servername, plus_feature, feature, load, secure_core = self.dashboard_service.set_individual_server( + servername, images_dict, servers, feature) if secure_core and only_secure_core: - tree_object.append(country_row, [images_dict["empty_pix"], servername, plus_feature, feature, load]) + tree_object.append(country_row, + [images_dict["empty_pix"], servername, plus_feature, feature, load]) elif not secure_core and not only_secure_core: - tree_object.append(country_row, [images_dict["empty_pix"], servername, plus_feature, feature, load]) - - + tree_object.append(country_row, + [images_dict["empty_pix"], servername, plus_feature, feature, load])